Spark未在DSE上使用DirectJoin

Question

我正在开发一个Spark流任务，该任务将流中的数据与Cassandra表连接起来。正如您在解释计划中看到的那样，没有使用直接加入。根据DSE文档，当(表大小*directJoinSizeAratio(>键的大小。在我的例子中，Table有数百万条记录，而密钥只有一条记录(表单流(，所以我希望使用Diret Join。表radice_polizza只有id_cod_poliza列作为分区jey。连接器版本：2.5.1。DSE版本：6.7.6。

*Project [id_cod_polizza#86L, progressivo#11, id3_numero_polizza#25, id3_cod_compagnia#21]
+- *SortMergeJoin [id_cod_polizza#86L], [id_cod_polizza#10L], Inner
:- *Sort [id_cod_polizza#86L ASC NULLS FIRST], false, 0
:  +- Exchange hashpartitioning(id_cod_polizza#86L, 200)
:     +- *Project [value#84L AS id_cod_polizza#86L]
:        +- *SerializeFromObject [input[0, bigint, false] AS value#84L]
:           +- Scan ExternalRDDScan[obj#83L]
+- *Sort [id_cod_polizza#10L ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(id_cod_polizza#10L, 200)
+- *Scan org.apache.spark.sql.cassandra.CassandraSourceRelation [id_cod_polizza#10L,progressivo#11,id3_numero_polizza#25,id3_cod_compagnia#21] ReadSchema: struct<id_cod_polizza:bigint,progressivo:string,id3_numero_polizza:string,id3_cod_compagnia:string>

这是我的代码：

var radice_polizza = spark
.read
.format("org.apache.spark.sql.cassandra")
.options(Map("table" -> "radice_polizza", "keyspace" -> "preferred_temp"))
.load().select(
"id_cod_polizza",
"progressivo",
"id3_numero_polizza",
"id3_cod_compagnia")
if(mode == LoadMode.DIFF){
val altered_data_df = altered_data.idCodPolizzaList.toDF("id_cod_polizza")
radice_polizza = altered_data_df.join(radice_polizza, Seq("id_cod_polizza"))
radice_polizza.explain()
}

强制直接加入是有效的。

radice_polizza = altered_data_df.join(radice_polizza.directJoin(AlwaysOn), Seq("id_cod_polizza"))

== Physical Plan ==
*Project [id_cod_polizza#58L, progressivo#11, id3_numero_polizza#25, id3_cod_compagnia#21]
+- DSE Direct Join [id_cod_polizza = id_cod_polizza#58L] preferred_temp.radice_polizza - Reading (id_cod_polizza, progressivo, id3_numero_polizza, id3_cod_compagnia) Pushed {}
+- *Project [value#56L AS id_cod_polizza#58L]
+- *SerializeFromObject [input[0, bigint, false] AS value#56L]
+- Scan ExternalRDDScan[obj#55L]

为什么不自动使用Direct Join？

Thnak你

Answer 1

当您使用DSE Analytics依赖关系开发应用程序时，DSE Direct Join会自动启用，该依赖关系是在DSE Analytics上运行作业时提供的。您需要为此指定以下依赖项，并且不要使用Spark Cassandra连接器：

<dependency>
<groupId>com.datastax.dse</groupId>
<artifactId>dse-spark-dependencies</artifactId>
<version>${dse.version}</version>
<scope>provided</scope>
</dependency>

如果在外部Spark上运行作业，则需要通过指定值为com.datastax.spark.connector.CassandraSparkExtensions的Spark配置属性spark.sql.extensions来显式启用直接联接。

我有一篇关于与卡桑德拉的加盟数据的长博客文章，解释了所有这些事情。