CODEWITHSUNDEEP
pythonapache-sparkpysparkspark-structured-streaming
espartian
espartian

Reputation: 45

Word Count with timestamp in Python

This example is extracted from Structured Streaming Programming Guide of Spark:

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split

spark = SparkSession \
        .builder \
        .appName("StructuredNetworkWordCount") \
        .getOrCreate()

# Create DataFrame representing the stream of input lines from connection to localhost:9999
   lines = spark \
     .readStream \
     .format("socket") \
     .option("host", "localhost") \
     .option("port", 9999) \
     .load()

# Split the lines into words
  words = lines.select(
    explode(
       split(lines.value, " ")
       ).alias("word"),
       lines.timestamp.alias('time')
)

# Generate running word count
 wordCounts = words.groupBy("word").count() #line to modify

# Start running the query that prints the running counts to the console
query = wordCounts \
      .writeStream \
      .outputMode("complete") \
      .format("console") \
      .start()

query.awaitTermination()

I need to create a table with every word and its input time. The output table should be like this:

+-------+--------------------+
|word   |              time  |
+-------+--------------------+
|   car |2021-12-16  12:21:..|
+-------+--------------------+

How can I do it? I think the line marked with "#line to modify" is only the line to modify.

Upvotes: 0

Views: 257

Answers (1)

Ged
Ged

Reputation: 18108

Try, something like this:

streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
  batchDF.persist()
  batchDF.write.format(...).save(...)  // location 1
  batchDF.write.format(...).save(...)  // location 2
  batchDF.unpersist()
}

You can do something like this:

writeStream
    .format("parquet")        // can be "orc", "json", "csv", etc.
    .option("path", "path/to/destination/dir")
    .start()

and make an external table to point, and set the path if needed yourself.

See https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#using-foreach-and-foreachbatch

Delta also writes to file location:

df.writeStream
  .format("delta")
  .outputMode("append")
  .option("checkpointLocation", "/delta/df/_checkpoints/etl-from-json")
  .start("/delta/df")

You may want to think about "complete".

Upvotes: 1

Related Questions