从Spark结构流中的Kafka消息中读取换行分隔的json

Question

我开发了一个Python Kafka生成器，它将多个json记录作为nd-json二进制字符串发送到一个Kafka主题。然后，我尝试用PySpark在Spark结构化流媒体中读取这些消息，如下所示：

events_df=select(from_json(col("value“）.cast（”string“）,schema).alias（”value“））

Answer 1

我设法用这种方式完成了我想要的工作，用换行符拆分全文字符串，然后将数组分解成要用模式解析的行：

    events = spark.readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "localhost:9092") \
        .option("subscribe", "quickstart-events") \
        .option("startingOffsets", "earliest")\
        .load()\
        .selectExpr("CAST(value AS STRING) as data")
    
    events = events.select(explode(split(events.data, '\n')))
    events = events.select(from_json(col("col"), event_schema).alias('value'))
    events = events.selectExpr('value.*')```