高级连接两个数据框 spark scala

Posted 2023-04-15

【中文标题】高级连接两个数据框 spark scala

【英文标题】：Advanced join two dataframe spark scala

【发布时间】：2018-12-16 16:18:40

【问题描述】：

我必须加入两个数据框。

示例： Dataframe1 是这样的

df1_col1      df1_col2
   a            ex1
   b            ex4
   c            ex2
   d            ex6
   e            ex3

数据框2

df2_col1      df2_col2
   1           a,b,c
   2           d,c,e
   3           a,e,c

在结果数据框中我想得到这样的结果

res_col1      res_col2       res_col3
    a           ex1             1
    a           ex1             3
    b           ex4             1
    c           ex2             1
    c           ex2             2
    c           ex2             3
    d           ex6             2
    e           ex3             2
    e           ex3             3

实现此连接的最佳方式是什么？

【参考方案1】：

我已经更新了下面的代码

val df1 = sc.parallelize(Seq(("a","ex1"),("b","ex4"),("c","ex2"),("d","ex6"),("e","ex3")))
val df2 = sc.parallelize(Seq(List(("1","a,b,c"),("2","d,c,e")))).toDF
df2.withColumn("df2_col2_explode", explode(split($"_2", ","))).select($"_1".as("df2_col1"),$"df2_col2_explode").join(df1.select($"_1".as("df1_col1"),$"_2".as("df1_col2")), $"df1_col1"===$"df2_col2_explode","inner").show

您只需要拆分值并通过爆炸生成多行，然后与其他数据框连接。

你可以参考这个链接，How to split pipe-separated column into multiple rows?

【讨论】：

感谢您的回答。我使用 spark sql 进行此连接。我在下面发布了我的解决方案。

@Sc0rpion 您正在对 Dataframe 和 RDD 执行连接，是否可以执行？

是的，但我更喜欢使用通过 rdd 提供的更高级别的抽象，而不是使用 rdd 本身。无论是 sql、dataframe 还是 datasets，都可以归结为 RDD。

【参考方案2】：

这个join我用spark sql，这里是部分代码；

df1.createOrReplaceTempView("temp_v_df1")
df2.createOrReplaceTempView("temp_v_df2")
val df_result = spark.sql("""select 
                    |   b.df1_col1 as res_col1, 
                    |   b.df1_col2 as res_col2, 
                    |   a.df2_col1 as res_col3  
                    |   from (select df2_col1, exp_col 
                    |         from temp_v_df2 
                    |        lateral view explode(split(df2_col2,",")) dummy as exp_col) a
                    |   join temp_v_df1 b on a.exp_col = b.df1_col1""".stripMargin)

【参考方案3】：

我使用 spark scala 数据框来实现你想要的输出。

val df1 = sc.parallelize(Seq(("a","ex1"),("b","ex4"),("c","ex2"),("d","ex6"),("e","ex3"))).toDF("df1_col1","df1_col2") 

val df2 = sc.parallelize(Seq((1,("a,b,c")),(2,("d,c,e")),(3,("a,e,c")))).toDF("df2_col1","df2_col2") 

df2.withColumn("_tmp", explode(split($"df2_col2", "\\,"))).as("temp").join (df1,$"temp._tmp"===df1("df1_col1"),"inner").drop("_tmp","df2_col2").show

愿望输出

+--------+--------+--------+
|df2_col1|df1_col1|df1_col2|
+--------+--------+--------+
|       2|       e|     ex3|
|       3|       e|     ex3|
|       2|       d|     ex6|
|       1|       c|     ex2|
|       2|       c|     ex2|
|       3|       c|     ex2|
|       1|       b|     ex4|
|       1|       a|     ex1|
|       3|       a|     ex1|
+--------+--------+--------+

根据您的要求重命名列。

这里是运行代码的截图

快乐哈杜啊啊啊啊啊啊啊啊啊啊啊啊啊啊啊啊啊啊啊啊啊啊啊