Spark不通过DSE使用DirectJoin

如何解决Spark不通过DSE使用DirectJoin

我正在开发一个Spark流任务，该任务将流中的数据与Cassandra Table连接起来。如您在解释计划中所见，未使用直接联接。 根据DSE的文档，当（表大小* directJoinSizeRatio）>键的大小时，将使用直接连接。 在我的情况下，表有数百万条记录，键仅是一条记录（流式传输），因此我希望使用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],0
      +- Exchange hashpartitioning(id_cod_polizza#10L,200)
         +- *Scan org.apache.spark.sql.cassandra.CassandraSourceRelation [id_cod_polizza#10L,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,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,false] AS value#56L]
         +- Scan ExternalRDDScan[obj#55L]

为什么不自动使用直接联接？

向你致敬

解决方法

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

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

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

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