I am trying to read some data from Kafka to see what is there.
I wrote
builder = SparkSession.builder\
.appName("PythonTest01")
spark = builder.getOrCreate()
# Subscribe to 1 topic
df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", config["kafka"]["bootstrap.servers"]) \
.option("subscribe", dataFlowTopic) \
.load()
# df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
df.printSchema()
df = df.first()
query = df \
.writeStream \
.outputMode('complete') \
.format('console') \
.start()
query.awaitTermination()
Unfortunately, it swears
pyspark.sql.utils.AnalysisException: Queries with streaming sources must be executed with writeStream.start();
what it wants and how to satisfy it?
If I remove first() it swears
Complete output mode not supported when there are no streaming aggregations on streaming DataFrames/Datasets;
I was to write
#df = df.first()
query = df \
.writeStream \
.outputMode('append') \
.format('console') \
.start()
query.awaitTermination()
which is printing not first, but last row and not terminates.