t1.py

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

# 创建SparkSession
spark = SparkSession \
    .builder \
    .appName("StructuredKafkaWordCount") \
    .getOrCreate()

# 读取Kafka数据
kafka_df = spark \
    .readStream \
    .format("my_kafka") \
    .option("my_kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "test") \
    .load()

# 提取数据中的value字段
kafka_values = kafka_df.selectExpr("CAST(value AS STRING)")

# 对数据执行flatMap和groupBy操作
words = kafka_values.select(
    explode(split(kafka_values.value, " ")).alias("word")
)
word_counts = words.groupBy("word").count()

# 输出结果
query = word_counts \
    .writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

query.awaitTermination()