scala - 当我们在连接列中有多个值时，如何在 Spark Scala 中应用连接

Question

我在两个文本文件中有数据

file 1:(patient id,diagnosis code)
+----------+-------+
|patient_id|diag_cd|
+----------+-------+
|         1|  y,t,k|
|         2|  u,t,p|
|         3|  u,t,k|
|         4|  f,o,k|
|         5|  e,o,u|
+----------+-------+

file2(diagnosis code,diagnosis description) Time T1
+-------+---------+
|diag_cd|diag_desc|
+-------+---------+
|      y|      yen|
|      t|      ten|
|      k|      ken|
|      u|      uen|
|      p|      pen|
|      f|      fen|
|      o|      oen|
|      e|      een|
+-------+---------+

文件 2 中的数据不是固定的并且不断变化，这意味着在任何给定时间点诊断代码 y 可以将诊断描述为日元，在其他时间点可以将诊断描述为十。例如下面

file2 at Time T2
+-------+---------+
|diag_cd|diag_desc|
+-------+---------+
|      y|      ten|
|      t|      yen|
|      k|      uen|
|      u|      oen|
|      p|      ken|
|      f|      pen|
|      o|      een|
|      e|      fen|
+-------+---------+

我必须在 spark 中读取这两个文件数据，并且只需要那些被诊断为 uen 的患者 id。它可以使用 spark sql 或 scala 来完成。

我试图在 spark-shell 中读取 file1。file1 中的两列是用竖线分隔的。

scala> val tes1 = sc.textFile("file1.txt").map(x => x.split('|')).filter(y => y(1).contains("u")).collect
tes1: Array[Array[String]] = Array(Array(2, u,t,p), Array(3, u,t,k), Array(5, e,o,u))

但是由于与诊断描述相关的诊断代码在 file2 中不是恒定的，因此必须使用连接条件。但是当 file1 中的 diag_cd 列有多个值时，我不知道如何应用连接。

任何帮助，将不胜感激。

标签： scalaapache-sparkapache-spark-sql

Answer 1

请在下面找到答案

//将file1读入数据框

val file1DF = spark.read.format("csv").option("delimiter","|")
.option("header",true)
.load("file1PATH")

//将file2读入数据框

val file2DF = spark.read.format("csv").option("delimiter","|")
.option("header",true)
.load("file2path")

//获取diag_desc的患者ID数据框作为uen

file1DF.join(file2DF,file1DF.col("diag_cd").contains(file2DF.col("diag_cd")),"inner")
.filter(file2DF.col("diag_desc") === "uen")
.select("patient_id").show