如何将流数据集写入 Cassandra？答案

【问题标题】：How to write streaming Dataset to Cassandra?如何将流数据集写入 Cassandra？
【发布时间】：2017-12-20 04:17:32
【问题描述】：

所以我有一个 Python Stream-sourced DataFrame df，它包含我想要放入带有 spark-cassandra-connector 的 Cassandra 表中的所有数据。我尝试过两种方式：

df.write \
    .format("org.apache.spark.sql.cassandra") \
    .mode('append') \
    .options(table="myTable",keyspace="myKeySpace") \
    .save() 

query = df.writeStream \
    .format("org.apache.spark.sql.cassandra") \
    .outputMode('append') \
    .options(table="myTable",keyspace="myKeySpace") \
    .start()

query.awaitTermination()

但是我不断收到此错误：

pyspark.sql.utils.AnalysisException: "'write' can not be called on streaming Dataset/DataFrame;

和

java.lang.UnsupportedOperationException: Data source org.apache.spark.sql.cassandra does not support streamed writing.

无论如何我可以将我的 Streamed DataFrame 发送到我的 Cassandra 表中吗？

【问题讨论】：

标签： apache-spark pyspark spark-cassandra-connector spark-structured-streaming

【解决方案1】：

目前在 Spark Cassandra 连接器中没有针对 Cassandra 的流式传输 Sink。您需要实现自己的 Sink 或等待它可用。

如果您使用的是 Scala 或 Java，则可以使用 foreach 运算符并使用 ForeachWriter，如 Using Foreach 中所述。

【讨论】：

有什么方法可以将我的 Streaming DataFrame 转换为非 Streaming 数据帧？
不，没有转化（至少据我所知没有）
你有 Java 的工作示例吗？看起来所有解决方案都来到CassandraConnector.withSessionDo，这需要Scala实现的特征；所以 Kotlin 或 Java 运气不好..
今天（我的意思是 2018 年）仍然如此吗？
在 Spark 2.3 及更高版本中，在创建自定义接收器时，看起来 Spark 不允许您在 addbatch 方法中在数据帧上调用 .write()。它抛出了 OP 共享的错误。有人知道这里的替代方案是什么吗？

【解决方案2】：

我知道这是一篇旧帖子，更新它以供将来参考。

您可以从流数据中将其作为批处理进行处理。如下所示

def writeToCassandra(writeDF, epochId):
 writeDF.write \
    .format("org.apache.spark.sql.cassandra") \
    .options(table="table_name", keyspace="keyspacename")\
    .mode("append") \
    .save()

query = sdf3.writeStream \
.trigger(processingTime="10 seconds") \
.outputMode("update") \
.foreachBatch(writeToCassandra) \
.start()

【讨论】：

它给了我错误，即无法找到数据源：org.apache.spark.sql.cassandra。我正在使用 --packages 选项传递包。