如何解决Spark Cassandra连接器+加入超时
我需要加入两个Spark数据帧，然后将结果返回给Hive。以下是数据框：
数据框1：Cassandra表-分区和聚类键：（ID，PART_NBR）
val df1 = spark.read.format("org.apache.spark.sql.cassandra")
.option("keyspace","mykeyspace")
.option("table","mytable")
.load
** Dataframe2：从另一个来源获得的键（属于上表中的ID列的分区键）的数据框-该表中不同键的数量约为15万个**
val df2 = spark.read
.format("jdbc")
.option("driver","com.mysql.jdbc.Driver")
.option("url","****")
.option("dbtable","table")
.option("user","username")
.option("password","password123")
.load()
val joinExpr = df1.col("ID") === df2.col("ID")
val res = df1.join(df2,joinExpr)
res.write.mode(SaveMode.Append).format("orc")
.saveAsTable("targetTable")
现在，此代码始终会导致“ com.datastax.oss.driver.api.core.servererrors.ReadFailureException：一致性为LOCAL_ONE的读取查询期间Cassandra失败（需要1个响应，但仅响应0个副本，1个失败）” 。
即使失败也将LOCAL_ONE更改为QUORUM。
我甚至尝试将密钥数据帧拆分为20个KEYS（数据帧中20个ID值）的批处理，然后与cassandra表联接-即使失败。
我什至尝试了IN子句，尽管它有效，但DBA限制了我们在加载Cassandra时运行它的操作，并导致CPU峰值。
在检查Cassandra DBA时，他们要求进行有针对性的查询，因为上述查询会导致较大的令牌范围扫描，并且会导致失败。但是，单独的有针对性的查询将导致与Cassandra的往返15万次（这需要花费几个小时才能完成），而且费用太高。
为什么会产生如此巨大的令牌范围扫描？我们该如何解决？我还有什么选择？
pom.xml依赖性
<scala.version>2.11.12</scala.version>
<spark.version>2.2.0</spark.version>
<dependency>
<groupId>com.datastax.spark</groupId>
<artifactId>spark-cassandra-connector_2.11</artifactId>
<version>2.5.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>${spark.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>${spark.version}</version>
<scope>provided</scope>
</dependency>
尝试了以下操作，但未进行直接联接。我有什么想念吗？
spark-submit --class ExampleCassandra --deploy-mode client --num-executors 15 --executor-memory 4g --driver-memory=1g --conf spark.sql.shuffle.partitions=25 --conf spark.executor.heartbeatInterval=100s --conf spark.sql.extensions=com.datastax.spark.connector.CassandraSparkExtensions --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --jars spark-sql_2.11-2.4.0.jar,spark-core_2.11-2.4.0.jar,spark-hive_2.11-2.4.0.jar,mysql-connector-java-8.0.18.jar,spark-cassandra-connector_2.11-2.5.1.jar ExampleCassandra-bundled-1.0-SNAPSHOT.jar
代码中印有= Spark.sparkContext.version = 2.4.0的火花版本
产生的计划
== Physical Plan ==
*(8) SortMergeJoin [item_nbr#31],[item_nbr#24],Inner
:- *(2) Sort [item_nbr#31 ASC NULLS FIRST],false,0
: +- Exchange hashpartitioning(item_nbr#31,25)
: +- *(1) Scan org.apache.spark.sql.cassandra.CassandraSourceRelation [item_nbr#31,planNum#32,strN#33,currTail#34,currTailTy#35,hor#36,prNbr#37,revSce#38,stckHnad#39] PushedFilters: [],ReadSchema: struct<item_nbr:int,planNum:int,strN:int,currTail:decimal(38,18),currTailTy:s...
+- *(7) Sort [item_nbr#24 ASC NULLS FIRST],0
+- Exchange hashpartitioning(item_nbr#24,25)
+- *(6) HashAggregate(keys=[item_nbr#21],functions=[])
+- Exchange hashpartitioning(item_nbr#21,25)
+- *(5) HashAggregate(keys=[item_nbr#21],functions=[])
+- *(5) Filter (NOT (trim(lower(item_nbr#21),None) = null) && isnotnull(cast(trim(item_nbr#21,None) as int)))
+- Generate explode(split(items#4,)),[item_nbr#21]
+- *(4) Project [items#4]
+- *(4) BroadcastHashJoin [planNum#0],[planNum#2],Inner,BuildRight
:- *(4) Scan JDBCRelation(( select planNum from QAMdPlans.Plan where plan_type = 'MBM' order by planNum desc ) t) [numPartitions=1] [planNum#0] PushedFilters: [*IsNotNull(planNum)],ReadSchema: struct<planNum:int>
+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0,int,false] as bigint)))
+- *(3) Scan org.apache.spark.sql.cassandra.CassandraSourceRelation [planNum#2,items#4] PushedFilters: [],ReadSchema: struct<planNum:int,items:string>
解决方法
问题在于版本2.0.5不能优化Dataframe的联接-如果您执行res.explain，则会看到Spark将执行从Cassandra读取所有数据的操作，然后对火花级别。优化的联接仅在RDD API中以leftJoinWithCassandraTable或joinWithCassandraTable的形式提供。
release of the Spark Cassandra Connector 2.5的情况已经改变，现在包括针对Dataframe API的优化联接（但是您需要启用Spark SQL扩展才能使其工作）。因此，您需要将连接器升级到2.5.latest（目前为2.5.1），或使用RDD API中的加入功能。
P.S。我最近wrote a detailed blog post介绍了同时使用Dataframe和RDD API与Spark中的Cassandra表中的数据进行有效连接的方法。