无法在PySpark应用程序中从Cassandra表加载数据

Question

无法从cassandra加载数据；据我了解，问题在于番石榴jar；尝试了不同版本的番石榴；无法识别预期的jar版本。(也替换为datastax共享jar )

版本- Scala 2.11.12，星火2.3.2.3.1.4.41-3罐，使用火花-卡桑德拉-连接器_2.11-2.3.2.jar，卡桑德拉-驱动-核心-3.0.jar，共用-配置-1.7.jar，java-驱动器阴影-番石榴-25.1-jre.jar或一个从番石榴罐版本19/24/31在火花提交。

电火花脚本&在以下内容之后提交：

Error:
  File "cass.py", line 6, in <module>
    data_df = ss.read.format("org.apache.spark.sql.cassandra").options(keyspace="xxxxx",table="xxxxx").load()
  File "/disk-3/hadoop/yarn/local/usercache/vdfidt1/appcache/application_1660663467107_0171/container_e68_1660663467107_0171_01_000001/pyspark.zip/pyspark/sql/readwriter.py", line 172, in load
  File "/disk-3/hadoop/yarn/local/usercache/vdfidt1/appcache/application_1660663467107_0171/container_e68_1660663467107_0171_01_000001/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1257, in __call__
  File "/disk-3/hadoop/yarn/local/usercache/vdfidt1/appcache/application_1660663467107_0171/container_e68_1660663467107_0171_01_000001/pyspark.zip/pyspark/sql/utils.py", line 63, in deco
  File "/disk-3/hadoop/yarn/local/usercache/vdfidt1/appcache/application_1660663467107_0171/container_e68_1660663467107_0171_01_000001/py4j-0.10.7-src.zip/py4j/protocol.py", line 328, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o103.load.

py4j.protocol.Py4JJavaError: An error occurred while calling o103.load.
: java.lang.NoSuchMethodError: com.google.common.base.Objects.firstNonNull(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
    at com.datastax.driver.core.policies.Policies$Builder.build(Policies.java:285)
    at com.datastax.driver.core.Cluster$Builder.getConfiguration(Cluster.java:1246)
    at com.datastax.driver.core.Cluster.<init>(Cluster.java:116)
    at com.datastax.driver.core.Cluster.buildFrom(Cluster.java:181)
    at com.datastax.driver.core.Cluster$Builder.build(Cluster.java:1264)
    at com.datastax.spark.connector.cql.DefaultConnectionFactory$.createCluster(CassandraConnectionFactory.scala:131)
    at com.datastax.spark.connector.cql.CassandraConnector$.com$datastax$spark$connector$cql$CassandraConnector$$createSession(CassandraConnector.scala:159)
    at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$8.apply(CassandraConnector.scala:154)
    at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$8.apply(CassandraConnector.scala:154)
    at com.datastax.spark.connector.cql.RefCountedCache.createNewValueAndKeys(RefCountedCache.scala:32)
    at com.datastax.spark.connector.cql.RefCountedCache.syncAcquire(RefCountedCache.scala:69)
    at com.datastax.spark.connector.cql.RefCountedCache.acquire(RefCountedCache.scala:57)
    at com.datastax.spark.connector.cql.CassandraConnector.openSession(CassandraConnector.scala:79)
    at com.datastax.spark.connector.cql.CassandraConnector.withSessionDo(CassandraConnector.scala:111)
    at com.datastax.spark.connector.rdd.partitioner.dht.TokenFactory$.forSystemLocalPartitioner(TokenFactory.scala:98)
    at org.apache.spark.sql.cassandra.CassandraSourceRelation$.apply(CassandraSourceRelation.scala:272)
    at org.apache.spark.sql.cassandra.DefaultSource.createRelation(DefaultSource.scala:56)
    at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:341)
    at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:239)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:164)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:282)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:750)

pyscript - cass.py

import pyspark
from pyspark.sql import SparkSession,SQLContext

ss = SparkSession.builder.appName("pyspark cassandra").getOrCreate()
data_df = ss.read.format("org.apache.spark.sql.cassandra")
    .options(keyspace="reg_idt1_vdf",table="veh")
    .load()
data_df.show()
ss.stop()

火花提交- cass.sh

function setJars() {
    i=0
    for file in /home/tst1/*.jar
    do
       JARS=${JARS},${file}
    done
}

setJars
    
spark-submit \
    --name 'pyspark cassandra connector' \
    --master yarn \
    --deploy-mode cluster \
    --driver-memory 2g \
    --executor-memory 2g \
    --num-executors 20 \
    --jars ${JARS} \
    --conf "spark.yarn.maxAppAttempts=1" \
    --conf "spark.cassandra.connection.host=xx.xx.xx.xx" \
    --conf "spark.cassandra.auth.username=xxxxx" \
    --conf "spark.cassandra.auth.password=xxxxxx" \
    --conf "spark.dynamicAllocation.enabled=true" \
    --conf "spark.dynamicAllocation.maxExecutors=100" \
    --conf "spark.dynamicAllocation.minExecutors=10"  \
    --conf "spark.executor.cores=2" \
    --conf "spark.dynamicAllocation.executorIdleTimeout=500s" \
    --conf "spark.authenticate=true" \
    $/home/tst1/cass.py
    exit 0

Answer 1

这个问题最有可能的原因是，与应用程序一起打包的组件中存在冲突。作为参考，Spark连接器v2.3.2的依赖项+对应版本列表为Versions.scala。

此外，不需要包含Cassandra驱动程序，因为连接器已经自动嵌入驱动程序。

我们的一般建议是使用spark-submit选项启动--packages。例如：

$ spark-submit --packages datastax:spark-cassandra-connector:2.3.2-s_2.11

这将自动下载使用连接器所需的所有依赖项。这还确保只包含兼容的版本。

如果你真的想要构建一个胖罐子，你可以下载Maven的“程序集”JAR，如果你的火花版本存在。连接器程序集JAR包含使用连接器所需的所有依赖项。干杯!

Answer 2

@Eric Ramirez，因为由于防火墙问题，我无法使用“-packages”，通过提供maven pom中提到的全部依赖项，最终能够从cassandra中提取数据。2.11-2.3.2.pom

非常感谢！！