如何处理从 Kafka 到 Cassandra 的 pySpark 结构化流

Question

我正在使用 pyspark 从 Kafka 获取数据并将其插入到 cassandra 中。我快到了，我只需要最后一步。

def Spark_Kafka_Receiver():

# STEP 1 OK!

    dc = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "000.00.0.240:9092") \
        .option("subscribe", "MyTopic") \
    .load()
    dc.selectExpr("CAST(key as STRING)", "CAST(value AS STRING) as msg")

# STEP 2 OK!

    dc.writeStream \
        .outputMode("append") \
        .foreachBatch(foreach_batch_function) \
        .start() \
        .awaitTermination()

# STEP 3 NEED HELP

def foreach_batch_function(df, epoch_id):
    Value = df.select(df.value)

    ???????

    # WRITE DATA FRAME ON CASSANDRA
    df.write \
        .format("org.apache.spark.sql.cassandra") \
        .mode('append') \
        .options(table=table_name, keyspace=keyspace) \
        .save()

所以我的值是这种格式的：

DataFrame[值：二进制]

我需要插入一些东西来打开我的值，将二进制文件放入其中，并创建一个具有正确格式的漂亮数据框来匹配数据库，并用它来执行我代码的最后一部分。

Answer 1

您不再需要使用 foreachBatch。你只需要升级到原生支持 Spark Structured Streaming 的 Spark Cassandra Connector 2.5，这样你就可以写：

dc.writeStream \
        .format("org.apache.spark.sql.cassandra") \
        .mode('append') \
        .options(table=table_name, keyspace=keyspace)
        .start() \
        .awaitTermination()

关于你问题的第二部分 - 如果你想将你的值转换成多列，你需要使用 from_json 函数，将架构传递给它。这是 Scala 中的示例，但 Python 代码应该非常相似：

val schemaStr = "id:int, value:string"
val schema = StructType.fromDDL(schemaStr)
val data = dc.selectExpr("CAST(value AS STRING)")
  .select(from_json($"value", schema).as("data"))
  .select("data.*").drop("data")

然后你可以通过writeStream

写入数据

如何处理从 Kafka 到 Cassandra 的 pySpark 结构化流

How to deal with pySpark structured streaming coming from Kafka to Cassandra

cassandra

apache-spark

pyspark

spark-cassandra-connector

spark-structured-streaming