如何解决PySpark和Cassandra
我对如何使用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
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。