spark不在dse上使用directjoin

v440hwme 于 2021-05-24 发布在 Spark

关注(0)|答案(1)|浏览(450)

我正在开发一个spark流任务，它将流中的数据与cassandra表连接起来。正如您在解释计划中看到的，没有使用直接连接。根据dse，当（表大小*directjoinsizeratio）>键大小时使用直接连接。在我的例子中，表有数百万条记录，键只有一条记录（formstreaming），所以我希望使用diretjoin。表radice\u polizza只有id\u cod\u polizza列作为分区jey。连接器version: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]

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

cassandra apache-spark spark-cassandra-connector spark-streaming dse

来源：https://stackoverflow.com/questions/64059794/spark-not-use-directjoin-over-dse

1条答案

按热度按时间

k97glaaz1#

当您使用在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 .
我有一篇关于与Cassandra联合数据的博文，解释了所有这些事情。

赞(0）回复(0）举报 2021-05-25

我来回答

spark不在dse上使用directjoin

1条答案

相关问题

热门标签

最新问答