使用 .net-spark 加载具有多个部分的固定位置文件

Posted 2023-04-13

【中文标题】使用 .net-spark 加载具有多个部分的固定位置文件

【英文标题】：Load fixed position file with multiple sections using .net-spark

【发布时间】：2021-09-27 18:24:47

【问题描述】：

我正在尝试使用 .net-spark 在 spark 中加载具有多个部分的固定位置文件。 这是文件的一个示例：

01Nikola Tesla                  [e-mail]                                                       [phone]
02Creations                                       
03Alternating current                              
03Tesla coil                                 
01Thomas Edison                 [e-mail]                                                      [phone]
02Creations                                        
03Lamp                                         
03Phonograph                                      
03General Eletric                                 
03Cinema 

所以基本上我们有一个交易所有者的标题，一个子标题说明下面是交易，最后是交易部分。交易行不包含对所有者的任何引用，所以是的，这很棘手。

正如@EdElliott 所建议的，这是我们应该如何查看RDD 中的数据（仅显示第一行，但目的是读取所有内容）：

inventor	email	phone	creations
Nikola Tesla	[e-mail]	[phone]	Alternating current

我猜这种文件格式现在不是很常见，但在巴西的大银行中仍然很常见。

找到this example for java，但它不处理部分。我相信我可以使用 UDF 来实现这一点，但同样，我不知道从哪里开始。在这里欣赏任何见解。

谢谢

【问题讨论】：

您能否举例说明您希望数据重新格式化后的样子？

当然，谢谢@EdElliott

【参考方案1】：

我通过了另一边。对解决方案并不感到自豪，主要是因为我正在使用 ToLocalIterator() 以及我将参数传递给 normalizedTransaction UDF 的方式。谢谢@EdElliott，your blog 帮了我很多忙。

不管怎样，就这样吧：

static Func<Column, Column> OnlyHeaders = Udf<string, bool>(
        line =>  line.Substring(0,2).Equals("01")
    );

    static Func<Column, Column> OnlyTransactions = Udf<string, bool>(
        line =>  line.Substring(0,2).Equals("03")
    );

    static Func<Column, Column> breakHeader =
        Udf<string, string[]>((line) => GetHeader(line) );

    static Func<Column, Column> breakTransaction =
        Udf<string, string[]>((line) => GetTransaction(line) );

    static void Main(string[] args)
    
        // Create a Spark session
        var spark = SparkSession
            .Builder()
            .AppName("FixedLenghtWithSectionsApp")
            .GetOrCreate();

        // Create initial DataFrame
        var rawDf = spark.Read().Schema("rawLine STRING").Text("resources/input.txt");
        rawDf = rawDf.WithColumn("rowNumber", Functions.MonotonicallyIncreasingId());
        rawDf.CreateOrReplaceTempView ("rawdata");

        var headersDf = GetHeadersDf(rawDf);
        headersDf.CreateOrReplaceTempView("headers");
        headersDf.Show();

        var transactionsDf = GetTransactionsDf(rawDf);
        transactionsDf.CreateOrReplaceTempView("transactions");
        transactionsDf.Show();
        var headerLines = headersDf.ToLocalIterator().ToList().Select( r => r.Get("rowNumber"));
        var columns = new StringBuilder();
        foreach(var h in headerLines)
            columns.Append($"h.ToString(),");
        
        
        var column = Functions.Lit(columns.ToString()).Alias("ids");

        Func<Column, Column, Column> normalizedTransaction = Udf<string, string,int>( (line, hLines) => 

                var ids = hLines.Split(",", StringSplitOptions.RemoveEmptyEntries).Select(s => int.Parse(s));
                var id = ids.Where(h => h < int.Parse(line) );
                if (id.Any())
                    return id.Max();
                return -1;
                
             );

        
        var inventionsDf = transactionsDf.Alias("one")
           .Select(
               Functions.Col("one.rowNumber"), 
               Functions.Col("invention"), 
               normalizedTransaction(Functions.Col("one.rowNumber").Cast("string") , column ).Alias("id")
            );

        inventionsDf = inventionsDf.Alias("one")
            .Join(
                headersDf.Alias("two")
                ,Functions.Col("one.id") == Functions.Col("two.rowNumber")
                ,"inner"
            );

        inventionsDf.Show();
        
        spark.Stop();
    



    private static DataFrame GetTransactionsDf(DataFrame rawDf)
    
        var transactionsRawsDf = rawDf
            .Select(rawDf["rowNumber"],breakTransaction(rawDf["rawLine"]).Alias("value"))
            .Where(OnlyTransactions(rawDf["rawLine"]));

        return transactionsRawsDf.Select(
            transactionsRawsDf["rowNumber"], 
            transactionsRawsDf.Col("value").GetItem(0))
            .ToDF("rowNumber", "invention");
        
    

    private static DataFrame GetHeadersDf(DataFrame rawDf)
    
        var headerRawsDf = rawDf
            .Select(rawDf["rowNumber"],breakHeader(rawDf["rawLine"]).Alias("value"))
            .Where(OnlyHeaders(rawDf["rawLine"]));
        
        
        return headerRawsDf.Select(
            headerRawsDf["rowNumber"], 
            headerRawsDf.Col("value").GetItem(0), 
            headerRawsDf.Col("value").GetItem(1),
            headerRawsDf.Col("value").GetItem(2))
            .ToDF("rowNumber", "inventor", "email", "phone");
    

    private static string[] GetHeader(string line)
    

        var columns = new List<string>();
            columns.Add(line.Substring(2,30));
            columns.Add(line.Substring(32,70));
            columns.Add(line.Substring(102,14));
            return columns.ToArray();
    

     private static string[] GetTransaction(string line)
    
        var columns = new List<string>();
            columns.Add(line.Substring(2,48));
            return columns.ToArray();