如何使用 Java SparkSql 实现内连接答案

【问题标题】：How to achieve inner join using Java SparkSql如何使用 Java SparkSql 实现内连接
【发布时间】：2020-02-19 16:10:51
【问题描述】：

我有两个数据集

数据集1

+-------+--------------------+  
|     id|                name|  
+-------+--------------------+  
|S703401|      Ryan P Cassidy|  
|S703401|Christopher J Mat...|  
|S703401|      Frank E LaSota|    
|S703401|      Ryan P Cassidy|  
|S703401|Anthony L Locricchio|  
|S703401|         Jason Monte|  
+-------+--------------------+

数据集2

+-------+------+  
|     id|   nic|  
+-------+------+    
|S703401|  RC82|  
|S703401|    NA|  
|S703401|   FL3|  
|S703401|  RC82|  
|S703401|    NA|  
|S703401|JM2080|  
+-------+------+

我想在 id 上加入他们，这样我就可以输出为

+-------+--------------------+-----------+  
|     id|                name|       nic |   
+-------+--------------------+-----------+  
|S703401|      Ryan P Cassidy|       RC82|  
|S703401|Christopher J Mat...|         NA|  
|S703401|      Frank E LaSota|        FL3|  
|S703401|      Ryan P Cassidy|       RC82|  
|S703401|Anthony L Locricchio|         NA|  
|S703401|         Jason Monte|     JM2080|  
+-------+--------------------+-----------+

我正在使用 java spark Dataset joined = dataset1.join(dataset2,"id"); 但他们我正在为所有行获取笛卡尔积，例如

+-------+--------------------+------+  
|     id |                 name|   nic|  
+-------+--------------------+------+  
|S703401|      Ryan P Cassidy|JM2080|  
|S703401|      Ryan P Cassidy|    NA|  
|S703401|      Ryan P Cassidy|  RC82|  
|S703401|      Ryan P Cassidy|   FL3|  
|S703401|      Ryan P Cassidy|    NA|  
|S703401|      Ryan P Cassidy|  RC82|  
|S703401|Christopher J Mat...|JM2080|  
|S703401|Christopher J Mat...|    NA|  
|S703401|Christopher J Mat...|  RC82|  
|S703401|Christopher J Mat...|   FL3|  
|S703401|Christopher J Mat...|    NA|  
|S703401|Christopher J Mat...|  RC82|  
|S703401|      Frank E LaSota|JM2080|  
|S703401|      Frank E LaSota|    NA|  
|S703401|      Frank E LaSota|  RC82|  
|S703401|      Frank E LaSota|   FL3|  
|S703401|      Frank E LaSota|    NA|  
|S703401|      Frank E LaSota|  RC82|  
|S703401|      Ryan P Cassidy|JM2080|  
|S703401|      Ryan P Cassidy|    NA|  
+-------+--------------------+------+

那么我在这里缺少什么？

【问题讨论】：

你在两个数据集中有多个相同的 id，所以很明显你会得到多对多的关系。你有什么逻辑可以决定你想要的输出吗？
我想在 dataset1 中添加 nic 值，其中 dataset1.id = dataset2.id，所以我只能得到 6 个输出

标签： java apache-spark apache-spark-sql

【解决方案1】：

不要使用,monotonically_increasing_id函数生成rowId。它会给出随机数，这将使您的 join 逻辑失败。因此，请使用 row_number 函数，它会在所有情况下为您提供正确的 rowId。

scala> import org.apache.spark.sql.expressions.Window

scala> df.show()
+-------+------------------+
|     id|              name|
+-------+------------------+
|S703401|      RyanPCassidy|
|S703401|ChristopherJMat...|
|S703401|      FrankELaSota|
|S703401|      RyanPCassidy|
|S703401|AnthonyLLocricchio|
|S703401|        JasonMonte|
+-------+------------------+


scala> df1.show()
+-------+------+
|     id|   nic|
+-------+------+
|S703401|  RC82|
|S703401|    NA|
|S703401|   FL3|
|S703401|  RC82|
|S703401|    NA|
|S703401|JM2080|
+-------+------+


scala> val w = Window.partitionBy(col("id")).orderBy(lit(1))

scala> df.withColumn("rw", row_number.over(w))
         .join(df1.withColumn("rw", row_number.over(w)), List("id", "rw"),"left")
         .drop("rw")
         .show(false)
+-------+------------------+------+
|id     |name              |nic   |
+-------+------------------+------+
|S703401|RyanPCassidy      |RC82  |
|S703401|ChristopherJMat...|NA    |
|S703401|FrankELaSota      |FL3   |
|S703401|RyanPCassidy      |RC82  |
|S703401|AnthonyLLocricchio|NA    |
|S703401|JasonMonte        |JM2080|
+-------+------------------+------+

【讨论】：