消息推送系统

import kafka

def send_message(topic, message):

producer = kafka.KafkaProducer(bootstrap_servers=['localhost:9092'])

producer.send(topic, value=message.encode('utf-8'))

producer.flush()

print("Message sent successfully!")

from pyspark.streaming import StreamingContext

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("UserBehaviorAnalysis").getOrCreate()

ssc = StreamingContext(spark.sparkContext, batchDuration=5)

def process_rdd(rdd):

df = spark.createDataFrame(rdd)

df.write.mode('append').format('jdbc').save(...)

ssc.socketTextStream("localhost", 9999).foreachRDD(process_rdd)

ssc.start()

ssc.awaitTermination()