scala - Spark 连接数组

Question

我是 spark 的新手(小时)，而且对 Scala 也相当缺乏经验。但是，我长期以来一直希望更加熟悉两者。

我有一个相当琐碎的任务。我有两个从两个 JSON 文件导入的数据框。一个带有 uuid,text,tag_ids ，另一个带有标签 id,term 我想生成一个新的 json 文件，我可以将其导入 solr 包含 uuid、text、tag_ids、tag_terms。

val text = spark.sqlContext.jsonFile("/tmp/text.js")
val tags = spark.sqlContext.jsonFile("/tmp/tags.js")



text.printSchema()

root
| -- uuid: string (nullable = true)
| -- tag_ids: array (nullable = true)
|    | -- element: string (contiansNull = true)
| -- text: string (nullable = true)

tags.printSchema()
root
| -- id: string (nullable = true)
| -- term: string (nullable = true)


#desired output  
+--------------------+------+---------+------------+
|                uuid| text | tag_ids |   tag_terms|  
+--------------------+------+---------+------------+    
|cf5c1f4c-96e6-4ca...| foo  |    [1,2]| [tag1,tag2]|      
|c9834e2e-0f04-486...| bar  |    [2,3]| [tag2,tag3]|   
+--------------------+--------------+--------------+

很难展示我一直在努力的一切。本质上，.join() 的问题在于 tag_ids 是一个数组。我可以 explode() tag_ids 并加入 tag_terms 但是重新组装成一个新的 df 来导出仍然超出了我的水平。

Answer 1

使用 explode 的解决方案:

val result = text
  .withColumn("tag_id", explode($"tag_ids"))
  .join(tags,  $"tag_id" === $"id")
  .groupBy("uuid", "tag_ids")
  .agg(first("text") as "text", collect_list("term") as "tag_terms")