PySpark 和 Cassandra

Question

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

这就是我想做的：

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

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

谢谢

Answer 1

当人们提到

pyspark-cassandra

时，他们大多会提到它，因为它公开了 Spark Cassandra 连接器 (SCC) 的 RDD 部分，而 SCC 本身并未公开该部分（对于 Python，它仅公开 Dataframe API）。

如何将 SCC 与 Astra 结合使用，在SCC 2.5.0 发布公告博客文章和文档中进行了很好的描述。 您可以使用以下命令启动 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

请注意禁用连续寻呼的标志 - 现在是必需的。

进程启动后，只需执行读取、转换和写入数据的 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