我用spark1.6尝试了下面的代码。
我已经编写了假设col1是字符串的解决方案。

步骤：

转换 DF1 到rdd[（字符串，字符串）]
组 DF2 由 COL1 加入结果 Step-1 与 Step-2 将结果Map到 Step-3 至 RDD[Row] 定义结构
创建新的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]|
+-----+--------+----------------------------------------------------------------+