I want to create a structured stream in databricks with a kafka source.
I followed the instructions as described here. My script seems to start, however I am unable to print/output something in the databricks notebook. The stream itsellf works fine and produces results and works (in databricks) when I use confluent_kafka, thus there seems to be a different issue I am missing:
The script seems "stuck" at "Running Command" / "Stream Initialising".
Any input is highly appreciated!
from pyspark.sql import functions as F
from pyspark.sql.types import *
# Define a data schema
schema = StructType() \
.add('PARAMETERS_TEXTVALUES_070_VALUES', StringType())\
.add('ID', StringType())\
.add('PARAMETERS_TEXTVALUES_001_VALUES', StringType())\
.add('TIMESTAMP', TimestampType())
df = spark \
.readStream \
.format("kafka") \
.option("host", "stream.xxx.com") \
.option("port", 12345)\
.option('kafka.bootstrap.servers', 'stream.xxx.com:12345') \
.option('subscribe', 'stream_test.json') \
.option("startingOffset", "earliest") \
.load()
df_word = df.select(F.col('key').cast('string'),
F.from_json(F.col('value').cast('string'), schema).alias("parsed_value"))
# Group by id and count
df_group = df_word.select('parsed_value.*')\
.groupBy('ID').count()
query = df_group \
.writeStream \
.outputMode("complete") \
.format("console") \
.start()
query.awaitTermination()
my stream output data looks like this:
"PARAMETERS_TEXTVALUES_070_VALUES":'something'
"ID":"47575963333908"
"PARAMETERS_TEXTVALUES_001_VALUES":12345
"TIMESTAMP": "2020-10-22T15:06:42.507+02:00"
To clarify: I am trying to print the components of query to the notebook to test the connection. There is no cell following this or previous to the above cell.
Thanks and stay safe.