Spark 不通过 DSE 使用 DirectJoin

Question

我正在开发一个 Spark 流式传输任务，它将流中的数据与 Cassandra 表连接起来。正如您在解释计划中看到的那样，不使用直接加入。 根据 DSE doc，当 (table size * directJoinSizeRatio) > size of keys 时使用 Direct Join。 在我的情况下，表有数百万条记录，键只有一条记录（表单流），所以我期待使用 Diret Join。 表 radice_polizza 只有 id_cod_polizza 列作为分区 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？

谢谢你

Answer 1

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

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

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

我有一个 long blog post 与 Cassandra 的联接数据，它解释了所有这些事情。