编写Spark结构流数据到Cassandra

Question

我想使用Pyspark API将结构流数据写入Cassandra。

我的数据流如下：

除非 - >卡夫卡 - >星火结构流 - >卡桑德拉;

我试过以下方式：

query = df.writeStream\
  .format("org.apache.spark.sql.cassandra")\
  .option("keyspace", "demo")\
  .option("table", "test")\
  .start()

但得到以下错误消息：“org.apache.spark.sql.cassandra”不支持流写入。

我尝试过另一种方法：[来源 - DSE 6.0 Administrator Guide]

query = df.writeStream\
   .cassandraFormat("test", "demo")\
   .start()

但得到异常：AttributeError：'DataStreamWriter'对象没有属性'cassandraFormat'

谁能让我知道如何进一步行动？

提前致谢。

Answer 1

升级DSE 6.0（最新版本）后，我可以将结构化流数据写入Cassandra。 [Spark 2.2＆Cassandra 3.11]

参考代码：

query = fileStreamDf.writeStream\
 .option("checkpointLocation", '/tmp/check_point/')\
 .format("org.apache.spark.sql.cassandra")\
 .option("keyspace", "analytics")\
 .option("table", "test")\
 .start()

DSE文档URL：Cajaxasopi

Answer 2

这个答案是为了向Cassandra写数据，而不是DSE（https://docs.datastax.com/en/dse/6.0/dse-dev/datastax_enterprise/spark/structuredStreaming.html）

对于Spark 2.4.0及更高版本，您可以使用foreachBatch方法，该方法允许您使用Spark Cassandra Connector提供的Cassandra批处理数据编写器将每个微批流传输查询的输出写入Cassandra：

which supports Structured Streaming for storing data

对于低于2.4.0的Spark版本，您需要实现一个foreach接收器。

import org.apache.spark.sql.cassandra._

df.writeStream
  .foreachBatch { (batchDF, _) => 
    batchDF
     .write
     .cassandraFormat("tableName", "keyspace")
     .mode("append")
     .save
  }.start

然后你可以使用foreach接收器如下：

import com.datastax.spark.connector.cql.CassandraConnector
import com.datastax.driver.core.querybuilder.QueryBuilder
import com.datastax.driver.core.Statement
import org.apache.spark.SparkConf
import org.apache.spark.sql.Row

class CassandraSink(sparkConf: SparkConf) extends ForeachWriter[Row] {
    def open(partitionId: Long, version: Long): Boolean = true

    def process(row: Row) = {
      def buildStatement: Statement =
        QueryBuilder.insertInto("keyspace", "tableName")
          .value("key", row.getAs[String]("value"))
      CassandraConnector(sparkConf).withSessionDo { session =>
        session.execute(buildStatement)
      }
    }

    def close(errorOrNull: Throwable) = Unit
}

Answer 3

除了以下情况，你在这里做的不多：

• 关注（和投票）

  df.writeStream
   .foreach(new CassandraSink(spark.sparkContext.getConf))
   .start

  。
• 实现所需的功能并自己打开PR。

除此之外，你可以创建使用corresponding JIRA并直接写。