Я использую Spark 2.3.0 с pyspark для подписки на поток 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 <span class=\"TitlePlannedWork\" >Planned Work</span>\n <br/>\n <a class=\"plannedWorkDetailLink\" onclick=ShowHide(184938);><b>Rockaways Long-Term Flood Protection | Until May 18, Mon to Fri, 6 AM to 10 AM & 3:30 PM to 8 PM<br clear=left>[A] No service <i>to/from</i> Rockaway Park-Beach 116 St</a><br/><br/><div id= 184938 class=\"plannedWorkDetail\" ><br></b>[A] trains that were scheduled to operate <i>to/from</i> Rockaway Park-Beach 116 St will operate <i>to/from</i> Euclid Av instead.<br><br>[S] No Rockaway Park Shuttle service at Broad Channel.<br><br>[S] Rockaway Park Shuttle operates, <i>approximately every 20 minutes,</i> between <b>Rockaway Park-Beach 116 St</b> and <b>Beach 90 St</b> and via the [A] <i>to/from</i> <b>Far Rockaway-Mott Av</b>.<br><br>For<b> Beach 90 St, Beach 98 St, Beach 105 St</b> and <b>Rockaway Park-Beach 116 St</b>, transfer to the [S] Rockaway Park Shuttle at Beach 67 St.<br><br><a href=http://www.mta.info/press-release/nyc-transit/nyc-transit-starting-subway-flood-protection-project-rockaways-next-month target=_blank><font color=#0000FF>Click here</font></a> for additional details on this Flood Protection Project.<br><br><table class=plannedworkTableStyle border=1 cellspacing=1 cellpadding=5 rules=none frame=box><td> [ad] <td><font size=1>This service change affects one or more ADA accessible stations. Please call 511 for help with planning<br>your trip. If you are deaf or hard of hearing, use your preferred relay service provider or the free 711 relay. </font></table><br><b><br></div></b><br/>\n <br/><br/>\n "
},
{
"status":"DELAYS",
"line":"BDFM",
"raw_text":"\n <span class=\"TitleDelay\">Delays</span>\n <span class=\"DateStyle\">\n Posted: 05/04/2018 4:29PM\n </span><br/><br/>\n [F] and [M] train service has resumed following an earlier incident involving a sick passenger at <STRONG>34 St-Herald Sq.</STRONG>\n <br/><br/>\n "
},
{
"status":"DELAYS",
"line":"G",
"raw_text":"\n <span class=\"TitleDelay\">Delays</span>\n <span class=\"DateStyle\">\n Posted: 05/04/2018 4:01PM\n </span><br/><br/>\n Court Sq-bound [G] trains are running with delays because of signal problems at<STRONG> Broadway</STRONG>.\n <br/><br/>\n "
},
{
"status":"GOOD SERVICE",
"line":"JZ",
"raw_text":null
},
{
"status":"DELAYS",
"line":"L",
"raw_text":"\n <span class=\"TitleDelay\">Delays</span>\n <span class=\"DateStyle\">\n Posted: 05/04/2018 4:27PM\n </span><br/><br/>\n [L] trains are running with delays in both directions because of a sick passenger at <STRONG>Canarsie-Rockaway Pkwy.</STRONG>\n <br/><br/>\n "
},
{
"status":"GOOD SERVICE",
"line":"NQR",
"raw_text":null
},
{
"status":"PLANNED WORK",
"line":"S",
"raw_text":"\n <span class=\"TitlePlannedWork\" >Planned Work</span>\n <br/>\n <a class=\"plannedWorkDetailLink\" onclick=ShowHide(184937);><b>Rockaways Long-Term Flood Protection | Until Friday May 18, 2018<br clear=left>[S] No Rockaway Park Shuttle service at Broad Channel - Take the [A] instead<br clear=left>[A] No rush hour service <i>to/from</i> Rockaway Park-Beach 116 St</a><br/><br/><div id= 184937 class=\"plannedWorkDetail\" ><br></b>[S] Rockaway Park Shuttle operates, <i>approximately every 20 minutes,</i> between <br><b>Rockaway Park-Beach 116 St</b> and <b>Beach 90 St</b> and via the [A] <i>to/from</i> <b>Far Rockaway-Mott Av</b>.<br><br>For <b>Broad Channel</b>, take the [A], transfer to the [S] Rockaway Park Shuttle at Beach 67 St.<br><br><b><i>Alternate travel note for Broad Channel:<br></i>Q52 </b>SBS, <b>Q53 </b>SBS, <b>QM16</b> and <b>QM17</b> service is also available at Cross Bay Blvd and Noel Rd.<br><br><a href=http://www.mta.info/press-release/nyc-transit/nyc-transit-starting-subway-flood-protection-project-rockaways-next-month target=_blank><font color=#0000FF>Click here</font></a> for additional details on this Flood Protection Project.<br><b><br></div></b><br/>\n <br/><br/>\n "
},
{
"status":"GOOD SERVICE",
"line":"SIR",
"raw_text":null
}
],
"timestamp":"5/4/2018 4:30:00 PM",
"oid":"5aecc363eb015557829c87c5"
}
Я не вижу каких-либо явных сообщений об ошибках или проблем на стороне потребителя, чтобы сказать, почему все мои значения отображаются как нули.
Есть ли способ легче понять, почему он неправильно анализирует значения?
Обновление:
Кажется, что кавычки могут быть частью проблемы. Для этого у меня есть объект словаря Python, который извлекается из коллекции mongodb, и каждая запись сбрасывается в строку с 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)
Не уверен, что есть лучший способ подготовить и отправить запись.