2

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

这就是我想要做的：

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

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

谢谢

4

1 回答 1

2

当人们提到pyspark-cassandra- 他们大多提到它，因为它暴露了 Spark Cassandra 连接器（SCC）的 RDD 部分，而 SCC 本身没有暴露（对于 Python，它只暴露 Dataframe API）。

SCC 2.5.0 发布公告博客文章和文档中很好地描述了如何将 SCC 与 Astra 一起使用。您使用以下命令启动 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
于 2020-08-26T06:41:51.950 回答