python - 使用 pyspark 和结构化流正确解析 kafka 流（使用当前模式处理获取所有空值）

问题描述

我正在使用带有 pyspark 的Spark 2.3.0来订阅 Kafka 流，目前正在尝试解析消息值，但获取每条记录的所有空值。

我的 kafka 版本是kafka_2.11-1.1.0代理版本0.10。

我正在运行具有以下内容的脚本：/opt/spark/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 ~/code/process/mta_processor.py

mta_processor.py看起来像这样：

import pyspark 
from pyspark import SparkContext
from pyspark.sql.session import SparkSession
from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
import bson

sc = SparkContext()
sc.setLogLevel("ERROR")
spark = SparkSession(sc)


df = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers","localhost:9092") \
    .option("subscribe", "mta-delays") \
    .option("startingOffsets", "earliest").load()

jsonschema = StructType().add("timestamp", StringType()) \
                         .add("timestamp_unix", IntegerType()) \
                         .add("oid", StringType()) \
                         .add("lines", ArrayType(StructType() \
                             .add("line", StringType()) \
                            .add("status", StringType()) \
                            .add("raw_text", StringType())))

mta_stream = df.select(from_json(col("value") \
                                .cast("string"), jsonschema) \
                                .alias("parsed_mta_values"))

mta_data = mta_stream.select("parsed_mta_values.*")


qry = mta_data.writeStream.outputMode("append").format("console").start()
qry.awaitTermination()

但结果都是空值：

Batch: 0
-------------------------------------------
+---------+--------------+----+-----+
|timestamp|timestamp_unix| oid|lines|
+---------+--------------+----+-----+
|     null|          null|null| null|
|     null|          null|null| null|
|     null|          null|null| null|
|     null|          null|null| null|
+---------+--------------+----+-----+

如果我只是用mta_data = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

当我获得内容的关键和价值时，那里有数据：

-------------------------------------------
Batch: 0
-------------------------------------------
+--------------------+--------------------+
|                 key|               value|
+--------------------+--------------------+
|5aecc1faeb0155502...|"{\"timestamp_uni...|
|5aecc254eb0155512...|"{\"timestamp_uni...|
|5aecc2b0eb0155545...|"{\"timestamp_uni...|
+--------------------+--------------------+

我从 kafka 生产者发送到我的 kafka 主题的数据如下所示。

{ "timestamp_unix":1525465800, "lines":[ { "status":"GOOD SERVICE", "line":"123", "raw_text":null }, { "status":"GOOD SERVICE", "line":"456", "raw_text":null }, { "status":"GOOD SERVICE", "line":"7", "raw_text":null }, { "status":"PLANNED WORK", "line":"ACE", "raw_text":"\n Planned Work\n \n <a class=\"plannedWorkDetailLink\" onclick=ShowHide(184938);>Rockaways Long-Term Flood Protection | Until May 18, Mon to Fri, 6 AM to 10 AM & 3:30 PM to 8 PM [A] No service to/from Rockaway Park-Beach 116 St</a> <div id= 184938 class=\"plannedWorkDetail\" > [A] trains that were scheduled to operate to/from Rockaway Park-Beach 116 St will operate to/from Euclid Av instead. [S] No Rockaway Park Shuttle service at Broad Channel. [S] Rockaway Park Shuttle operates, approximately every 20 minutes, between Rockaway Park-Beach 116 St and Beach 90 St and via the [A] to/from Far Rockaway-Mott Av. For Beach 90 St, Beach 98 St, Beach 105 St and Rockaway Park-Beach 116 St, transfer to the [S] Rockaway Park Shuttle at Beach 67 St. <a href=http://www.mta.info/press-release/nyc-transit/nyc-transit-starting-subway-flood-protection-project-rockaways-next-month target=_blank>Click here</a> for additional details on this Flood Protection Project. <table class=plannedworkTableStyle border=1 cellspacing=1 cellpadding=5 rules=none frame=box><td> [ad] <td>This service change affects one or more ADA accessible stations. Please call 511 for help with planning your trip. If you are deaf or hard of hearing, use your preferred relay service provider or the free 711 relay. </table> </div> \n \n " }, { "status":"DELAYS", "line":"BDFM", "raw_text":"\n Delays\n \n  Posted: 05/04/2018  4:29PM\n \n [F] and [M] train service has resumed following an earlier incident involving a sick passenger at 34 St-Herald Sq.\n \n " }, { "status":"DELAYS", "line":"G", "raw_text":"\n Delays\n \n  Posted: 05/04/2018  4:01PM\n \n Court Sq-bound [G] trains are running with delays because of signal problems at Broadway.\n \n " }, { "status":"GOOD SERVICE", "line":"JZ", "raw_text":null }, { "status":"DELAYS", "line":"L", "raw_text":"\n Delays\n \n  Posted: 05/04/2018  4:27PM\n \n [L] trains are running with delays in both directions because of a sick passenger at Canarsie-Rockaway Pkwy.\n \n " }, { "status":"GOOD SERVICE", "line":"NQR", "raw_text":null }, { "status":"PLANNED WORK", "line":"S", "raw_text":"\n Planned Work\n \n <a class=\"plannedWorkDetailLink\" onclick=ShowHide(184937);>Rockaways Long-Term Flood Protection | Until Friday May 18, 2018 [S] No Rockaway Park Shuttle service at Broad Channel - Take the [A] instead [A] No rush hour service to/from Rockaway Park-Beach 116 St</a> <div id= 184937 class=\"plannedWorkDetail\" > [S] Rockaway Park Shuttle operates, approximately every 20 minutes, between Rockaway Park-Beach 116 St and Beach 90 St and via the [A] to/from Far Rockaway-Mott Av. For Broad Channel, take the [A], transfer to the [S] Rockaway Park Shuttle at Beach 67 St. Alternate travel note for Broad Channel: Q52 SBS, Q53 SBS, QM16 and QM17 service is also available at Cross Bay Blvd and Noel Rd. <a href=http://www.mta.info/press-release/nyc-transit/nyc-transit-starting-subway-flood-protection-project-rockaways-next-month target=_blank>Click here</a> for additional details on this Flood Protection Project. </div> \n \n " }, { "status":"GOOD SERVICE", "line":"SIR", "raw_text":null } ], "timestamp":"5/4/2018 4:30:00 PM", "oid":"5aecc363eb015557829c87c5" }

我没有在消费者方面看到任何明显的错误消息或问题来说明为什么我的所有值都显示为空值。

反正有没有更容易弄清楚为什么它不能正确解析值？

更新：

似乎引号可能是问题的一部分。对于它的价值，我有一个从 mongodb 集合中提取的 python 字典对象，每个记录都被转储到字符串中prepared_record = json.dumps(record)，然后发送：

producer = KafkaProducer(bootstrap_servers='localhost:9092', 
                     value_serializer=lambda v: json.dumps(v).encode('utf-8'))

producer.send(MTA_DELAYS_IN_KAFKA_TOPIC, key=obj_key.encode(), value=prepared_record).get(timeout=30)

不确定是否有更好的方法来准备和发送记录。

标签： pythonapache-sparkpysparkapache-kafka

python - 使用 pyspark 和结构化流正确解析 kafka 流（使用当前模式处理获取所有空值）

问题描述

解决方案

推荐阅读