scala - 如何创建第三列的其他数据框包含由第一个数据框的 id 过滤的第二个数据框的数组？

Question

我有两个带有人员地址的数据框。一个表包含可靠来源地址和其他不可信来源地址。我们想知道地址是真人的可证明性如何。

我正在使用 Spark 1.5

我有两个数据框:

DF1

COL1  | COL2 
00001 | Street 1    
00001 | Street 2    
00002 | Street 1    
00002 | Street 2    
00002 | Street 1    

DF2

COL1  | COL2   
00001 | Street 1    
00001 | Street 2    
00001 | Street 2    
00001 | Street 2    
00002 | Street 1    
00002 | Street 2    
00002 | Street 1   

我的问题是如何转换为这个数据框/ map /其他？我的意思是转换成类似下面的代码。

COL1  | COL2        | COL3 (Array or Vector)   
00001 | Street 1    | [00001 | Street 1, 00001 | Street 2, 00001 | Street 2, 00001 | Street 2]    
00001 | Street 2    | [00001 | Street 1, 00001 | Street 2, 00001 | Street 2, 00001 | Street 2]    
00002 | Street 1    | [00002 | Street 1, 00002 | Street 2, 00002 | Street 1]    
00002 | Street 2    | [00002 | Street 1, 00002 | Street 2, 00002 | Street 1]   
00002 | Street 1    | [00002 | Street 1, 00002 | Street 2, 00002 | Street 1]   

最后一张表只是一个例子，我需要加入独立于文件格式表的两个数据帧。我们需要的是处理第三张表的数据进行统计。

Answer 1

我已经使用 spark 1.6 尝试了以下代码。

我已经假设 COL1 是一个字符串来编写解决方案。

步骤:

1. 将 DF1 转换为 RDD[(String,String)]
2. 按COL1对DF2进行分组>
3. 将 Step-1 的结果与 Step-2 合并
4. 将 Step-3 的结果映射到 RDD[Row]
5. 定义结构
6. 创建一个新的DataFrame

代码:

val aggRDD = df1.rdd.map(r=>         // Converting DF1 to RDD[(String,String)]
  (r.getString(0),                   //Mapping COL1
    r.getString(1))                  //Mapping COL2
).join(                              //Joining DF1 with grouped DF2 
  df2.rdd.groupBy(r=>r.getString(0)) // Grouping DF2 by COL1
).map(r=>                            // r -> (String, (String, Iterable[Row]))
  Row.fromSeq(Seq(                   // Creating a Row Object
    r._1,                            // Setting COL1
    r._2._1,                         // Setting COL2
    r._2._2.map(i=>                  // Converting the Row object to String
      i.getString(0)+"|"+i.getString(1) // Extracting COL1 and COL2 Value
    )
  ))
)

// Defining the Struct for the new DataFrame
val struct = new StructType(Array(
    StructField("COL1",StringType,nullable = true),
    StructField("COL2",StringType,nullable = true),
    StructField("Array",
    ArrayType(StringType,containsNull = true),nullable = true)
))    

// Creating a new DataFrame from RDD[Row]
val df = sq.createDataFrame(aggRDD,struct)

df.show(truncate=false)

+-----+--------+----------------------------------------------------------------+
|COL1 |COL2    |Array                                                           |
+-----+--------+----------------------------------------------------------------+
|00002|Street 1|[00002|Street 1, 00002|Street 2, 00002|Street 1]                |
|00002|Street 2|[00002|Street 1, 00002|Street 2, 00002|Street 1]                |
|00002|Street 1|[00002|Street 1, 00002|Street 2, 00002|Street 1]                |
|00001|Street 1|[00001|Street 1, 00001|Street 2, 00001|Street 2, 00001|Street 2]|
|00001|Street 2|[00001|Street 1, 00001|Street 2, 00001|Street 2, 00001|Street 2]|
+-----+--------+----------------------------------------------------------------+