如何在pyspark dataframe中读取scylladb表

Question

--packages datastax:spark-cassandra-connector:2.3.0-s_2.11 , 

我的Spark-version=2.3.1，Scala-version-2.11.8。

**First Approach**
from pyspark import SparkConf
from pyspark import SparkContext
from pyspark.sql import SparkSession
conf = SparkConf().set("spark.cassandra.connection.host","192.168.0.118")
sc = SparkContext(conf = conf)
spark=SparkSession.builder.config(conf=conf).appName('FinancialRecon').getOrCreate()
sqlContext =SQLContext(sc)
data=spark.read.format("org.apache.spark.sql.cassandra").options(table="datarecon",keyspace="finrecondata").load().show()

产生的错误：

文件“/usr/local/spark/python/lib/pyspark.zip/pyspark/sql/readwriter.py”，第172行，加载文件“/usr/local/spark/python/lib/py4j-0.10.7-src.zip/pyspark/py4j/java_gateway.py”中，第1257行，调用文件“/usr/local/spark/py4j/lib/py4j/java_gateway.py”中，第63行，解码器文件“4JJavaError：调用O43.load时出错。：java.lang.ClassNotFoundException：org.apache.Spark.logging在Spark2.0中被删除。请在org.apache.Spark.sql.execution.datasources.datasource$.lookupdatasource(datasource.scala:646)在org.apache.Spark.sql.dataframereader.load(dataframereader.scala:190)在org.apache.Spark.sql.dataframereader.load(dataframereader.scala:164)在sun.reflect.nativeMethodAccessorimpl.invoke0（原生方法）在在java.lang.reflect.Method.invoke(Method.java:498)在py4j.reflection.methodinvoker.invoke(methodinvoker.java:244)在py4j.reflection.methodinvoker.java:244)在py4j.reflection.reflection.engine.invoke(reflectionengine.java:357)在py4j.gateway.invoke(gateway.java:282)在8）在java.lang.thread.run（thread.java：748）处由：j引起Ava.lang.noClassDeffounderRoR:org/apache/spark/logging at java.lang.ClassLoader.DefineClass1(Native Method)，at java.lang.ClassLoader.DefineClass1(classLoader.java:763)，at java.security.secureClassLoader.defineClass1(secureClassLoader.java:142)，at java.net.urlClassLoader.java:467)，at:362)在java.security.accessController.doprivileged（本机方法）在java.net.urlClassLoader.findclass(URLClassLoader.java:361)在java.lang.ClassLoader.loadClass(classLoader.java:424)在sun.misc.launcher$appClassLoader.loadClass(launcher.java:349)在java.lang.ClassLoader.loadClass(classLoader.java:411)在）在org.apache.spark.sql.execution.datasources.datasource$$anonfun$23$$anonfun$apply$15.apply(datasource.scala:618)在scala.util.try$.apply(try.scala:192)在org.apache.spark.sql.execution.datasources.datasource$$$anonfun$23在org.apache.spark.execution.datasources.datasource$$$anonfun$23在scala.util.try.datasources(try.scala:618)在OURCE$.lookupDataSource(datasource.scala:618)...13个原因是:java.lang.ClassNotFoundException:org.apache.spark.logging在java.net.urlClassLoader.findClass(urlClassLoader.java:381)在java.lang.ClassLoader.loadClass(classLoader.java:424)在sun.misc.launcher$appClassLoader.loadClass(launcher.java:349)在

我使用过的另一个处理方法是：

data=sc.read.format("org.apache.spark.sql.cassandra").options(table="datarecon",keyspace="finrecondata").load().show()

为此我得到：

AttributeError：“SparkContext”对象没有属性“read”

data=sqlContext.read.format("org.apache.spark.sql.cassandra").options(table="datarecon",keyspace="finrecondata").load().show()

对此，我得到了与第一种方法相同的错误。

请指教这是scylla火花连接器的问题或一些火花库的问题，以及如何解决它。

Answer 1

按照以下步骤操作：

1.使用packages行运行spark-shell。要配置默认的Spark配置，使用--conf传递键值对，在我的示例中，scylla主机是172.17.0.2

bin/spark-shell --conf spark.cassandra.connection.host=172.17.0.2 --packages datastax:spark-cassandra-connector:2.3.0-s_2.11

2.在SparkContext、SparkSession、RDD和DataFrame上启用特定于Cassandra的函数：

import com.datastax.spark.connector._
import org.apache.spark.sql.cassandra._

val rdd = sc.cassandraTable("my_keyspace", "my_table")

scala> rdd.collect().foreach(println)
CassandraRow{id: 1, name: ash}