How to transform structured streams with PySpark?

Question

This seems like it should be obvious, but in reviewing the docs and examples, I'm not sure I can find a way to take a structured stream and transform using PySpark.

For example:

from pyspark.sql import SparkSession

spark = (
    SparkSession
    .builder
    .appName('StreamingWordCount')
    .getOrCreate()
)

raw_records = (
    spark
    .readStream
    .format('socket')
    .option('host', 'localhost')
    .option('port', 9999)
    .load()
)

# I realize there's a SQL function for upper-case, just illustrating a sample
# use of an arbitrary map function
records = raw_records.rdd.map(lambda w: w.upper()).toDF()

counts = (
    records
    .groupBy(records.value)
    .count()
)

query = (
    counts
    .writeStream
    .outputMode('complete')
    .format('console')
    .start()
)
query.awaitTermination()

This will throw the following exception:

Queries with streaming sources must be executed with writeStream.start

However, if I remove the call to rdd.map(...).toDF() things seem to work fine.

Seems as though the call to rdd.map branched execution from the streaming context and causes Spark to warn that it was never started?

Is there a "right" way to apply map or mapPartition style transformations using Structured Streaming and PySpark?

Answer 1

Every transformation that is applied in Structured Streaming has to be fully contained in Dataset world - in case of PySpark it means you can use only DataFrame or SQL and conversion to RDD (or DStream or local collections) are not supported.

If you want to use plain Python code you have to use UserDefinedFunction.

from pyspark.sql.functions import udf

@udf
def to_upper(s)
    return s.upper()

raw_records.select(to_upper("value"))

See also Spark Structured Streaming and Spark-Ml Regression