CODEWITHSUNDEEP
apache-sparkpysparkapache-kafkaspark-structured-streaming
abkf12
abkf12

Reputation: 221

How to read from Kafka and print out records to console in Structured Streaming in pyspark?

I'm using Spark 2.4.3, Scala 2.11.8, Java 1.8 and using this spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.3 data_stream.py for job submit.

The following is the code that gives an exception (see below):

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

df.printSchema()

schema = StructType([
    StructField("crime_id", StringType(), True),
    StructField("original_crime_type_name", StringType(), True),
    StructField("report_date", StringType(), True),
    StructField("call_date", StringType(), True),
    StructField("offense_date", StringType(), True),
    StructField("call_time", StringType(), True),
    StructField("call_date_time", StringType(), True),
    StructField("disposition", StringType(), True),
    StructField("address", StringType(), True),
    StructField("city", StringType(), True),
    StructField("state", StringType(), True),
    StructField("agency_id", StringType(), True),
    StructField("address_type", StringType(), True),
    StructField("common_location", StringType(), True)
])

kafka_df_string = df.selectExpr("CAST(value AS STRING)")
service_table = kafka_df_string.select(psf.from_json(psf.col('value'), schema).alias("SERVICE_CALLS"))
distinct_table = service_table.select(psf.col('crime_id'))
query = distinct_table.writeStream.format("console").start()
query.awaitTermination()

When I run the code it gives the following error:

Traceback (most recent call last):
  File "/Users/PycharmProjects/data-streaming-project/streaming/data_stream.py", line 55, in <module>
    run_spark_job(spark)
  File "/Users/PycharmProjects/data-streaming-project/streaming/data_stream.py", line 39, in run_spark_job
    distinct_table = service_table.select(psf.col('crime_id'))
  File "/Users/dev/spark-2.3.0-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/sql/dataframe.py", line 1202, in select
  File "/Users/dev/spark-2.3.0-bin-hadoop2.7/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py", line 1160, in __call__
  File "/Users/dev/spark-2.3.0-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/sql/utils.py", line 69, in deco
pyspark.sql.utils.AnalysisException: u"cannot resolve '`crime_id`' given input columns: [SERVICE_CALLS];;\n'Project ['crime_id]\n+- AnalysisBarrier\n      +- Project [jsontostructs(StructField(crime_id,StringType,true), StructField(original_crime_type_name,StringType,true), StructField(report_date,StringType,true), StructField(call_date,StringType,true), StructField(offense_date,StringType,true), StructField(call_time,StringType,true), StructField(call_date_time,StringType,true), StructField(disposition,StringType,true), StructField(address,StringType,true), StructField(city,StringType,true), StructField(state,StringType,true), StructField(agency_id,StringType,true), StructField(address_type,StringType,true), StructField(common_location,StringType,true), value#21, Some(America/Los_Angeles)) AS SERVICE_CALLS#23]\n         +- Project [cast(value#8 as string) AS value#21]\n            +- StreamingRelationV2 org.apache.spark.sql.kafka010.KafkaSourceProvider@5c2220e0, kafka, Map(startingOffsets -> earliest, subscribe -> service-calls, kafka.bootstrap.servers -> localhost:9092), [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13], StreamingRelation DataSource(org.apache.spark.sql.SparkSession@6faadbc2,kafka,List(),None,List(),None,Map(startingOffsets -> earliest, subscribe -> service-calls, kafka.bootstrap.servers -> localhost:9092),None), kafka, [key#0, value#1, topic#2, partition#3, offset#4L, timestamp#5, timestampType#6]\n"

Any help is appreciated!

EDIT

Now that I appended the .select("SERVICE_CALLS.*") in the code, I'm getting a ton of error in awaitTermination line.

pyspark.sql.utils.StreamingQueryException: u'org.apache.spark.sql.kafka010.KafkaMicroBatchReader.createDataReaderFactories()Ljava/util/List;\n=== Streaming Query ===\nIdentifier: [id = 6c33d7c9-3f9c-428d-aece-66f94315545a, runId = a6a41d93-b470-4636-b292-172ab13c36c7]\nCurrent Committed Offsets: {}\nCurrent Available Offsets: {KafkaV2[Subscribe[service-calls]]: {"service-calls":{"0":1414}}}\n\nCurrent State: ACTIVE\nThread State: RUNNABLE\n\nLogical Plan:\nProject [crime_id#25]\n+- Project [SERVICE_CALLS#23.crime_id AS crime_id#25, SERVICE_CALLS#23.original_crime_type_name AS original_crime_type_name#26, SERVICE_CALLS#23.report_date AS report_date#27, SERVICE_CALLS#23.call_date AS call_date#28, SERVICE_CALLS#23.offense_date AS offense_date#29, SERVICE_CALLS#23.call_time AS call_time#30, SERVICE_CALLS#23.call_date_time AS call_date_time#31, SERVICE_CALLS#23.disposition AS disposition#32, SERVICE_CALLS#23.address AS address#33, SERVICE_CALLS#23.city AS city#34, SERVICE_CALLS#23.state AS state#35, SERVICE_CALLS#23.agency_id AS agency_id#36, SERVICE_CALLS#23.address_type AS address_type#37, SERVICE_CALLS#23.common_location AS common_location#38]\n   +- Project [jsontostructs(StructField(crime_id,StringType,true), StructField(original_crime_type_name,StringType,true), StructField(report_date,StringType,true), StructField(call_date,StringType,true), StructField(offense_date,StringType,true), StructField(call_time,StringType,true), StructField(call_date_time,StringType,true), StructField(disposition,StringType,true), StructField(address,StringType,true), StructField(city,StringType,true), StructField(state,StringType,true), StructField(agency_id,StringType,true), StructField(address_type,StringType,true), StructField(common_location,StringType,true), value#21, Some(America/Los_Angeles)) AS SERVICE_CALLS#23]\n      +- Project [cast(value#8 as string) AS value#21]\n         +- StreamingExecutionRelation KafkaV2[Subscribe[service-calls]], [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13]\n'

with

java.lang.AbstractMethodError: org.apache.spark.sql.kafka010.KafkaMicroBatchReader.createDataReaderFactories()Ljava/util/List;

Upvotes: 1

Views: 2775

Answers (1)

Jacek Laskowski
Jacek Laskowski

Reputation: 74669

There are a couple of things here actually. One is a kind of typo while the other is a more severe.

service_table dataframe has just a single column SERVICE_CALLS after you do kafka_df_string.select(psf.from_json(psf.col('value'), schema).alias("SERVICE_CALLS")) so you cannot service_table.select(psf.col('crime_id')) as the crime_id column does not actually exist. That was easy, wasn't it? :)

The more severe issue is with spark-submit (which is from /Users/dev/spark-2.3.0-bin-hadoop2.7 directory) while the --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.3 uses 2.4.3 for the Spark version. They simply don't match and hence the exception:

java.lang.AbstractMethodError: org.apache.spark.sql.kafka010.KafkaMicroBatchReader.createDataReaderFactories()Ljava/util/List;

Please use --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 (for Spark 2.3.0) to match your spark-submit and you should be fine.

Upvotes: 3

Related Questions