使用分隔符拆分字符串并在 in 语句中使用它

Question

我们有两个表，首先包含每次运行中每个任务的代码。第二个表包含每次运行中所有任务的代码。如何运行基于分隔符拆分第二个表中的列的 spark sql 并在第一个表中使用它 in 和 IN 语句

表格如下所示

表 f1

+-------+-----+--+
| runid | tid |  |
+-------+-----+--+
| 1a    | cb4 |  |
| 1a    | hb5 |  |
| 1a    | hb6 |  |
| 1b    | gh6 |  |
| 1b    | gh7 |  |
| 1c    | kl9 |  |
+-------+-----+--+

表 f2

+-------+-------------+
| runid |     tid     |
+-------+-------------+
| 1a    | cb4,hb5,hb6 |
| 1b    | gh6,gh7,gh8 |
+-------+-------------+

我尝试过拆分，但它似乎不起作用，regexp_extract似乎没有帮助

select e.* from f1 e inner join
f2 a 
on e.runid=a.runid where e.runid in ('1a',
'1b') and e.tid in (select split(a.tid, '[,]') from f2)

期望的输出将是

+-------+-----+
| runid | tid |
+-------+-----+
| 1a    | cb4 |
| 1a    | hb5 |
| 1a    | hb6 |
| 1b    | gh6 |
| 1b    | gh7 |
+-------+-----+

就火花sql而言，我是菜鸟。任何帮助将不胜感激

Answer 1

将 lateral view 与 explode 一起使用，每行获取一个 tid，然后将其用于join。

with exploded_f2 as
(select runid,tid,expl_tid
 from f2
 lateral view explode(split(tid,',')) tbl as expl_tid
) 
select f1.*
from f1
join exploded_f2 f2 on f1.tid = f2.expl_tid

Answer 2

这是另一个版本：

val df3 = df2.flatMap {x => x.getString(1).split(",")
             .map(y => (x.getString(0),y))}.toDF("runid","tid2")
df3.show()
+-----+----+
|runid|tid2|
+-----+----+
|   1a| cb4|
|   1a| hb5|
|   1a| hb6|
|   1b| gh6|
|   1b| gh7|
|   1b| gh8|
+-----+----+

然后加入 df1 和 df3

df1.join(df3, "runid").select($"runid", $"tid").distinct.show(false)
+-----+---+
|runid|tid|
+-----+---+
|1a   |hb5|
|1b   |gh7|
|1b   |gh6|
|1a   |hb6|
|1a   |cb4|
+-----+---+ 

Answer 3

将数据加载为管道分隔的平面文件

from pyspark.sql.functions import *
from pyspark.sql.types import *
schema=StructType([StructField("runid",StringType()),StructField("tid",StringType())])
f1=spark.read.format("csv").schema(schema).option("header","true").option("delimiter","|").load("c:/tmp/f1.csv")
f2=spark.read.format("csv").schema(schema).option("header","true").option("delimiter","|").load("c:/tmp/f2.csv")

使用逗号作为分隔符分解，并将分解列重命名为 TID

f2_alter=(f2.withColumn("tid_explode",explode(split(f2.tid,"[,]")))).select("runid",col("tid_explode").alias("tid"))

在 Runid 和 TID 上做一个联接

df2=f1.join(f2_alter,["runid","tid"]).show()

+-----+---+
|runid|tid|
+-----+---+
|   1a|cb4|
|   1a|hb5|
|   1a|hb6|
|   1b|gh6|
|   1b|gh7|
+-----+---+