CODEWITHSUNDEEP
apache-sparkpysparkapache-kafkaapache-spark-sqlspark-streaming
tardis
tardis

Reputation: 1370

Spark dataframe lose streaming capability after accessing Kafka source

I use Spark 2.4.3 and Kafka 2.3.0. I want to do Spark structured streaming with data coming from Kafka to Spark. In general it does work in the test mode but since I have to do some processing of the data (and do not know another way to do) the Spark data frames do not have the streaming capability anymore.

#!/usr/bin/env python3

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json
from pyspark.sql.types import StructField, StructType, StringType, DoubleType

# create schema for data
schema = StructType([StructField("Signal", StringType()),StructField("Value", DoubleType())])

# create spark session
spark = SparkSession.builder.appName("streamer").getOrCreate()

# create DataFrame representing the stream
dsraw = spark.readStream \
  .format("kafka").option("kafka.bootstrap.servers", "localhost:9092") \
  .option("subscribe", "test")
print("dsraw.isStreaming: ", dsraw.isStreaming)

# Convert Kafka stream to something readable
ds = dsraw.selectExpr("CAST(value AS STRING)")
print("ds.isStreaming: ", ds.isStreaming)

# Do query on the converted data
dsQuery = ds.writeStream.queryName("ds_query").format("memory").start()
df1 = spark.sql("select * from ds_query")
print("df1.isStreaming: ", df1.isStreaming)

# convert json into spark dataframe cols
df2 = df1.withColumn("value", from_json("value", schema))
print("df2.isStreaming: ", df2.isStreaming)

The output is:

dsraw.isStreaming:  True
ds.isStreaming:  True
df1.isStreaming:  False
df2.isStreaming:  False

So I lose the streaming capability when I create the first dataframe. How can I avoid it? How do I get a streaming Spark dataframe out of a stream?

Upvotes: 0

Views: 546

Answers (1)

cronoik
cronoik

Reputation: 19550

It is not recommend to use the memory sink for production applications as all the data will be stored in the driver.

There is also no reason to do this, except for debugging purposes, as you can process your streaming dataframes like the 'normal' dataframes. For example:

import pyspark.sql.functions as F

lines = spark.readStream.format("socket").option("host", "XXX.XXX.XXX.XXX").option("port", XXXXX).load()

words = lines.select(lines.value)

words = words.filter(words.value.startswith('h'))

wordCounts = words.groupBy("value").count()

wordCounts = wordCounts.withColumn('count', F.col('count') + 2)

query = wordCounts.writeStream.queryName("test").outputMode("complete").format("memory").start()

In case you still want to go with your approach: Even if df.isStreaming tells you it is not a streaming dataframe (which is correct), the underlying datasource is a stream and the dataframe will therefore grow with each processed batch.

Upvotes: 1

Related Questions