apache-spark - PySpark 和 Cassandra

Question

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

使用安全的 bundle.zip (astra datastax) 将 pyspark 连接到远程 datastax 数据库

从数据库中检索我的信息以将其用于机器学习目的，将数据检索到 python 数据框或其他东西中。

这就是我想要做的，但是我看过这么多帖子，但没有一个完全有效，我不想直接使用 pyspark shell，如果可能的话，我想在某些代码编辑器中用 python 代码完成所有操作，我的意思是，无 Spark 端子内。
谢谢

Answer 1

当人们提到 pyspark-cassandra - 他们大多提到它是因为它公开了 Spark Cassandra Connector (SCC) 的 RDD 部分，而 SCC 本身并未公开(对于 Python，它仅公开数据帧 API)。
如何在 Astra 中使用 SCC 在 SCC 2.5.0 release announcement blog post 中有很好的描述。 ，并在 documentation .您可以使用以下命令启动 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