两个数据集之间的 Java + Apache Spark 内连接

Question

请教关于如何使用 Apache Spark 和 Java 实现内部连接的小问题。

我有这段非常简单的代码。

final Dataset<Row> dataSetTableLeft = getDatasetForLeft();
final Dataset<Row> dataSetTableRight = getDatasetForRight();
final Dataset<Row> dataSetTableResult = dataSetTableLeft.join(dataSetTableRight);

dataSetTableLeft.show();
dataSetTableRight.show();
dataSetTableResult.show();

第一张表dataSetTableLeft.show ，长这样，很直观。

+----------+-----+
|      time|label|
+----------+-----+
|1637020800|    0|
|1637107200|    0|
|1637193600|    0|
|1637280000|    0|
|1637366400|    0|
|1637452800|    0|
+----------+-----+"

第二张表dataSetTableRight，长这样，也很直观。

+----------+-----+
|      time|label|
+----------+-----+
|1637193600|    1|
|1637280000|    2|
|1637366400|    1|
+----------+-----+"

我想要实现的是这样一个结果表，它是一种内连接。

+----------+-----+
|      time|label|
+----------+-----+
|1637020800|    0|
|1637107200|    0|
|1637193600|    1|
|1637280000|    2|
|1637366400|    1|
|1637452800|    0|
+----------+-----+"

不幸的是，我没有看到任何这样的功能。 innerjoin()

因此，我正在尝试某种组合

dataSetTableLeft.unionAll(dataSetTableRight);
dataSetTableLeft.crossJoin(dataSetTableRight);

到目前为止没有任何运气。

实现内连接的正确方法是什么？

Answer 1

内连接只会为您提供两个数据集中存在的行。您在这里想要的是保留dataSetTableLeft 中的所有行，并在time 匹配时从dataSetTableRight 检索label 值。

为此，只需在匹配时使用左连接和coalesce 函数从第二个数据集中获取label 值。

以下是提供数据的完整工作示例：

import static org.apache.spark.sql.functions.*;

dataSetTableLeft.show();
//+----------+-----+
//|      time|label|
//+----------+-----+
//|1637020800|    0|
//|1637107200|    0|
//|1637193600|    0|
//|1637280000|    0|
//|1637366400|    0|
//|1637452800|    0|
//+----------+-----+

dataSetTableRight.show();
//+----------+-----+
//|      time|label|
//+----------+-----+
//|1637193600|    1|
//|1637280000|    2|
//|1637366400|    1|
//+----------+-----+


dataSetTableLeft.alias("df1").join(
        dataSetTableRight.alias("df2"),
        dataSetTableLeft.col("time").equalTo(dataSetTableRight.col("time")),
        "left_outer"
).select(
        dataSetTableLeft.col("time"),
        expr("coalesce(df2.label, df1.label)").alias("label")
).show();

//+----------+-----+
//|      time|label|
//+----------+-----+
//|1637020800|    0|
//|1637107200|    0|
//|1637193600|    1|
//|1637280000|    2|
//|1637366400|    1|
//|1637452800|    0|
//+----------+-----+