如何在流批处理流连接中定义连接条件？

Question

我使用的是SPARK-SQL-2.4.1V和Java1.8。和Kafka版本SPARK-SQL-KAFKA-0-10_2.11_2.4.3。

 Dataset<Row> streamingDs  = //read from kafka topic
 Dataset<Row> staticDf=  //read from oracle meta-data table.


Dataset<Row> joinDf = streamingDs.as("c").join(staticDf.as("i") ,
                      "c.code = i.industry_code"
                      );

Dataset<Row> joinDf = streamingDs.as("c").join(staticDf.as("i") ,
                      "c.code = i.industry_code",
                      "inner"
                      );

这会产生以下错误：

类型Dataset中的方法join（Dataset,String）不适用于参数（Dataset,String,String）

Answer 1

tl；drc.code=i.industry_code被认为是要联接的列的名称（而不是联接表达式）。

将代码更改为：

streamingDs.as("c").join(staticDf.as("i")) // INNER JOIN is the default
  .where("c.code = i.industry_code")