PySpark和Cassandra

编程问答 2022-08-14

问题描述

我对如何使用pyspark和cassandra连接器感到非常困惑。一些帖子说这只是使用pyspark中的SparkContext和SparkSession的问题，而其他帖子则说那些不起作用，我必须使用pyspark-cassandra。有人可以告诉我将datastax远程cassandra数据库与pyspark连接的正确方法是什么吗？

这就是我想要做的：

  • 使用安全的bundle.zip（astra datastax）将pyspark连接到远程datastax数据库
  • 从数据库中检索我的信息，以将其用于机器学习目的，将数据检索到python数据框等中。

这是我想要做的，但是我看到了太多的帖子，但没有一个完全起作用，我不想直接使用pyspark shell，如果可能的话，我想在某些代码编辑器中使用python代码完成所有操作，我是说，火花端子内没有。

谢谢

解决方法

当人们提到pyspark-cassandra时-人们大多提到它，因为它公开了Spark Cassandra Connector（SCC）的RDD部分，而SCC本身并未公开（对于Python，它仅公开了Dataframe API）。

如何在SCC 2.5.0 release announcement blog post和documentation中描述如何在Astra中使用SCC。您可以使用以下命令启动pyspark（您可以在代码内指定用户名，密码和其他参数，但--packages除外，在命令行上不需要）

pyspark --packages com.datastax.spark:spark-cassandra-connector_2.11:2.5.1\ 
  --files path_to/secure-connect-test.zip \
  --conf spark.cassandra.connection.config.cloud.path=secure-connect-test.zip \
  --conf spark.cassandra.auth.username=UserName \
  --conf spark.cassandra.auth.password=Password \
  --conf spark.dse.continuousPagingEnabled=false

请注意该标志禁用了连续分页-如this post中所述，它现在是必需的。

过程开始后，只需执行用于读取，转换和写入数据的Spark命令：

>>> from pyspark.sql.functions import col

# read data
>>> data = park.read.format("org.apache.spark.sql.cassandra")\
   .options(table="t2",keyspace="test").load()
>>> data.count()
5
>>> data.show(5,truncate = False)
+---+-----------------------+
|id |tm                     |
+---+-----------------------+
|4  |2020-06-23 10:37:25.825|
|3  |2020-06-23 10:37:25.754|
|5  |2020-06-23 10:37:25.852|
|1  |2020-06-23 10:37:25.701|
|2  |2020-06-23 10:37:25.726|
+---+-----------------------+

# generate new data frame
>>> data2 = data.select((col("id") + 10).alias("id"),col("tm"))
>>> data2.show()
+---+--------------------+
| id|                  tm|
+---+--------------------+
| 13|2020-06-23 10:37:...|
| 14|2020-06-23 10:37:...|
| 15|2020-06-23 10:37:...|
| 11|2020-06-23 10:37:...|
| 12|2020-06-23 10:37:...|
+---+--------------------+

# write the data
>>> data2.write.format("org.apache.spark.sql.cassandra")\
  .options(table="t2",keyspace="test").mode("append").save()

# check that data is written
>>> spark.read.format("org.apache.spark.sql.cassandra")\
  .options(table="t2",keyspace="test").load().count()
10
apache-sparkcassandrapysparkspark-cassandra-connector