Spark streaming: get the max values

Question

Hi I am triying to get the most repeated values from a stream data.

In order to do this I have the following code:

import os
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.functions import regexp_extract, col

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.2.0,org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.0 pyspark-shell'

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

df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", 'localhost:9092') \
    .option("subscribe", 'twitter') \
    .option("startingTimestamp", 1000) \
    .option("startingOffsets", "earliest") \
    .load()

ds = df \
    .selectExpr("CAST(value AS STRING)", "timestamp") \
    .select(regexp_extract(col('value'), '#(\w+)', 1).alias('hashtags'), 'timestamp')

df_group = ds.withWatermark("timestamp", "5 seconds") \
    .groupBy(
        'timestamp',
        'hashtags'
    ).agg(
        F.count(col('hashtags')).alias('total')
    )

query = df_group  \
    .writeStream  \
    .outputMode("append")  \
    .format("console")  \
    .option("truncate", "False")  \
    .start()
    
query.awaitTermination()

The idea is to process a batch of 5 seconds, and show when each batch is processed the current top hashtags most used.

The main idea was using this code without group by timestamp, but I got an error, about that if ds doesn't use timestamp then df_group doesn't use outputMode("append"), and I want to show the update.

Is this possible, how can I do it?

Thanks.

Spark streaming: get the max values

0 Answers