Spark Cassandra 连接器 3.0.0 - 如何启用 DirectJoin - Java

Question

我有 cassandra 3.11.9、spark 3.0.1 和 Spark-cassandra-connector 3.0.0（依赖项）。我正在尝试使用 SCC 3.0.0 的直接连接，但似乎当我在下面的数据集上使用连接时，我得到了 Spark 的广播哈希连接。

Dataset<Row> metlistinitial = sp.read().format("org.apache.spark.sql.cassandra")
            .options(new HashMap<String, String>() {
                {
                    put("keyspace", "mdb");
                    put("table", "experiment");
                }
            })
            .load().select(col("experimentid"), col("description"))
            .join(dfexplist,"experimentid")
            .filter(col("description").notEqual("Unidentified"));
metlistinitial.explain();

== Physical Plan ==
*(1) Project [experimentid#6, description#7]
+- *(1) BroadcastHashJoin [experimentid#6], [experimentid#4], Inner, BuildRight
   :- *(1) Project [experimentid#6, description#7]
   :  +- *(1) Filter NOT (description#7 = Unidentified)
   :     +- BatchScan[experimentid#6, description#7] Cassandra Scan: mdb.experiment
 - Cassandra Filters: []
 - Requested Columns: [experimentid,description]
   +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true])), [id=#19]
  +- LocalTableScan [experimentid#4]

我应该做些什么来启用与 cassandra 表的直接连接吗？现在连接大约需要 8 分钟，我想看看直接连接是否会更快。

Answer 1

刚刚找到了！看来我只需要通过添加来激活连接器

.config("spark.sql.extensions","com.datastax.spark.connector.CassandraSparkExtensions")

在火花配置中。惊人的表现。现在只需要8秒！

Answer 2

如果您能够使用数据集的直接连接，您可以帮助我吗？我一直在努力让它在 Java 中工作，并且没有太多关于此的示例