<jupyter_start><jupyter_text>1. Import Spark Libraries<jupyter_code>import findspark
findspark.init("/usr/local/spark-3.2.0-bin-hadoop3.2")
from IPython.display import display, clear_output
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.streaming import DataStreamReader
from pyspark.sql import functions as f
from pyspark.conf import SparkConf
from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql.functions import *
import html
import numpy as np
import json
import matplotlib.pyplot as plt
import pandas as pd
import seaborn as sns
pd.options.display.max_columns = None
pd.options.display.max_rows = 250
pd.options.display.max_colwidth = 150
sns.set(color_codes=True)
from pyspark.ml import PipelineModel
from pyspark.sql.functions import udf
from pyspark.sql.types import *
import warnings
warnings.filterwarnings('ignore')<jupyter_output><empty_output><jupyter_text>2. Build SparkSession<jupyter_code>spark = SparkSession.builder\
.master("local[*]") \
.appName("ml") \
.config("spark.memory.fraction","0.8") \
.config("spark.executor.memory","8g") \
.config("spark.driver.memory","8g") \
.config("spark.sql.hive.filesourcePartitionFileCacheSize", "621440000") \
.config("spark.sql.sources.bucketing.maxBuckets", "100000") \
.config("spark.sql.shuffle.partitions", "2000") \
.config("spark.driver.maxResultSize","2g") \
.config("spark.shuffle.file.buffer","64k") \
.config("spark.scheduler.listenerbus.eventqueue.capacity", "1000") \
.config("spark.broadcast.blockSize", "8m") \
.config("spark.sql.autoBroadcastJoinThreshold", "-1") \
.config('spark.jars.packages', 'org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.0')\
.getOrCreate()<jupyter_output><empty_output><jupyter_text>3. Load Trained Model<jupyter_code>MODEL_PATH = "/home/jovyan/work/lr_model/lr_sakarya_twitter_sentiment_analysis_model.pkl"
sentiment_model = PipelineModel.load(MODEL_PATH)<jupyter_output><empty_output><jupyter_text>4. Build Spark Stream<jupyter_code>df_stream = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "172.16.31.10:9092") \
  .option("subscribe", "tweet") \
  .option("startingOffsets", "latest") \
  .option("kafka.request.timeout.ms", "60000") \
  .option("kafka.session.timeout.ms", "30000") \
  .option("kafkaConsumer.pollTimeoutMs", "5120") \
  .option("failOnDataLoss", "true") \
  .option("fetchOffset.numRetries", "5") \
  .load()<jupyter_output><empty_output><jupyter_text>5. Convert Data Coming from Kafka<jupyter_code>df_schema = StructType() \
        .add("createdAt", IntegerType()) \
        .add("name", StringType()) \
        .add("likeCount", IntegerType()) \
        .add("quoteCount", IntegerType()) \
        .add("replyCount", IntegerType()) \
        .add("retweetCount", IntegerType()) \
        .add("text", StringType())

df_stream_cast = df_stream.selectExpr("CAST(value AS STRING)").select(from_json(col("value"), df_schema).alias("values"))
df = df_stream_cast.select("values.*")<jupyter_output><empty_output><jupyter_text>6. Clean Data<jupyter_code>@udf
def html_unescape(s: str):
    if isinstance(s, str):
        return html.unescape(s)
    return s


def clean_data(df: DataFrame):
    url_regex = r"((https?|ftp|file):\/{2,3})+([-\w+&@#/%=~|$?!:,.]*)|(www.)+([-\w+&@#/%=~|$?!:,.]*)"
    email_regex = r"[\w.-]+@[\w.-]+\.[a-zA-Z]{1,}"
    user_regex = r"(@\w{1,15})"
    
    return (
        df
        
        .withColumn("original_text", f.col("text"))
        
        .withColumn("text", f.regexp_replace(f.col("text"), url_regex, ""))
        .withColumn("text", f.regexp_replace(f.col("text"), email_regex, ""))
        .withColumn("text", f.regexp_replace(f.col("text"), user_regex, ""))
        .withColumn("text", f.regexp_replace(f.col("text"), "#", " "))
        
        .withColumn("text", html_unescape(f.col("text")))
        
        .withColumn("text", f.regexp_replace(f.col("text"), "[^a-zA-Z']", " "))
        .withColumn("text", f.regexp_replace(f.col("text"), " +", " "))
        .withColumn("text", f.trim(f.col("text")))
        
    
        .filter(f.col("text") != "").na.drop(subset="text")
    )

df_clean=clean_data(df)<jupyter_output><empty_output><jupyter_text>7. Predict Sentiment Tweets<jupyter_code>raw_sentiment = sentiment_model.transform(df_clean)<jupyter_output><empty_output><jupyter_text>8. Real Time Predicted Sentiment Data<jupyter_code>sentiment = raw_sentiment.select(
    "createdAt", "name", "likeCount","quoteCount","replyCount",
    "retweetCount","text", "original_text",
    f.col("prediction").alias("user_sentiment")
)<jupyter_output><empty_output><jupyter_text>9. Stream Aggregation Queries<jupyter_code>negative_sentiment_count = (
    sentiment.filter("user_sentiment == 0")
    .select(f.col("user_sentiment").alias("negative_sentiment"))
    .agg(f.count("negative_sentiment"))
)

positive_sentiment_count = (
    sentiment.filter("user_sentiment == 1")
    .select(f.col("user_sentiment").alias("positive_sentiment"))
    .agg(f.count("positive_sentiment"))
)

average_sentiment = sentiment.agg(f.avg("user_sentiment"))<jupyter_output><empty_output><jupyter_text>10. Real Time Predicted Data to Sink Kafka<jupyter_code>stream_writer = (
    sentiment
    .selectExpr("to_json(struct(*)) AS value")
    .writeStream
    .queryName("emotion_sentiment")
    .format("kafka") \
    .option("kafka.bootstrap.servers", "172.16.31.10:9092")
    .option("topic", "sentiment_topic")
    .option("checkpointLocation", "/home/jovyan/work/logs")
)

query = stream_writer.start()
query.name<jupyter_output><empty_output><jupyter_text>11. Real Time Agg Queries<jupyter_code>if df_clean.isStreaming:
    display(query.lastProgress)
stream_writer2 = (
    positive_sentiment_count
    .selectExpr("to_json(struct(*)) AS value")
    .writeStream
    .outputMode("complete")
    .queryName("emotion_sentiment_agg")
    .format("kafka") \
    .option("kafka.bootstrap.servers", "172.16.31.10:9092")
    .option("topic", "sentiment_agg_topic")
    .option("checkpointLocation", "/home/jovyan/work/logs2")
)

query2 = stream_writer2.start()
df_stream_cast2 = df_stream.selectExpr("CAST(value AS STRING)", "timestamp") \
.select(from_json(col("value"), df_schema) \
        .alias("values"),"timestamp")
df2 = df_stream_cast2.select("values.*", "timestamp")

df_clean2=clean_data(df2)

raw_sentiment2 = sentiment_model.transform(df_clean2)


sentiment2 = raw_sentiment2.select(
    "createdAt", "name", "likeCount","quoteCount","replyCount",
    "retweetCount","text", "original_text", "timestamp",
    f.col("prediction").alias("user_sentiment")
)
sentiment_count = \
  sentiment2 \
    .withWatermark("timestamp", "2 minutes") \
    .groupBy(
      "user_sentiment", "timestamp") \
    .count()
stream_writer3 = (
    sentiment_count
    .selectExpr("to_json(struct(*)) AS value")
    .writeStream
    .outputMode("append")
    .queryName("emotion_sentiment_agg2")
    .format("kafka") \
    .option("kafka.bootstrap.servers", "172.16.31.10:9092")
    .option("topic", "sentiment_agg_groupby_topic")
    .option("checkpointLocation", "/home/jovyan/work/logs3")
)

query3 = stream_writer3.start()
words = df_clean2.withColumn('word', f.explode(f.split(f.col('original_text'), ' '))) \
    .withWatermark('timestamp', '2 minutes')
wordCounts = words.groupBy('word', 'timestamp').count()
stream_writer4 = (
    wordCounts
    .selectExpr("to_json(struct(*)) AS value")
    .writeStream
    .outputMode("append")
    .queryName("emotion_sentiment_word_agg")
    .format("kafka") \
    .option("kafka.bootstrap.servers", "172.16.31.10:9092")
    .option("topic", "sentiment_word_agg_topic")
    .option("checkpointLocation", "/home/jovyan/work/logs4")
)

query4 = stream_writer4.start()
words2 = df_clean2.withColumn('word', f.explode(f.split(f.col('original_text'), ' ')))
wordCounts2 = words2.groupBy('word').count()
stream_writer5 = (
    wordCounts2
    .selectExpr("to_json(struct(*)) AS value")
    .writeStream
    .outputMode("complete")
    .queryName("emotion_sentiment_word_agg3")
    .format("kafka") \
    .option("kafka.bootstrap.servers", "172.16.31.10:9092")
    .option("topic", "sentiment_word_agg_topic2")
    .option("checkpointLocation", "/home/jovyan/work/logs5")
)

query5 = stream_writer5.start()
query.isActive
query2.isActive
query3.isActive
query4.isActive
query5.isActive
df_clean.isStreaming
query.explain()
display(query.lastProgress)
query.awaitTermination()
spark.streams.awaitAnyTermination()
query.stop()
query2.stop()
query3.stop()
query4.stop()
query5.stop()
spark.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.17649114631873253, 'fraction_numerical': 0.024580615097856476, 'mean_word_length': 4.9206896551724135, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 6, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '343892', 'license': 'MIT', 'n_tokens_mistral': 3188, 'n_tokens_neox': 3069, 'n_words': 401, 'path': 'pyspark_twitter_streaming_with_kafka.ipynb', 'repo_name': 'muhammeedsari/pyspark_stream', 'size': '8596'}