Spark Streaming的核心功能及其示例PySpark代码

时间：2025-01-23 07:10:19

Spark Streaming是Apache Spark中用于实时流数据处理的模块。以下是一些常见功能的实用PySpark代码示例：

  1. 基础流处理：从TCP套接字读取数据并统计单词数量
from pyspark import `SparkContext
from pyspark.streaming import StreamingContext

# 创建SparkContext和StreamingContext
sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, 1)  # 1秒的批处理间隔

# 创建一个DStream，从TCP源读取数据
lines = ssc.socketTextStream("localhost", 9999)

# 对每一行数据进行分词，映射为(word, 1)的键值对，然后按单词统计数量
words = lines.flatMap(lambda line: line.split(" "))
word_counts = words.map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)

# 打印每个RDD中的前10个元素
word_counts.pprint()

# 启动流计算
ssc.start()
# 等待流计算结束
ssc.awaitTermination()

在上述代码中：

  • sc 是 SparkContext ，用于与Spark集群交互。
  • ssc 是 StreamingContext ，定义了批处理间隔。
  • lines 是一个 DStream ，从指定的TCP套接字读取数据。
  • words 对每行数据进行分词， word_counts 统计每个单词出现的次数。
  • pprint 方法打印每个批次的前10个元素。
  1. 使用窗口函数
from pyspark import SparkContext
from pyspark.streaming import StreamingContext

sc = SparkContext("local[2]", "WindowedWordCount")
ssc = StreamingContext(sc, 1)

lines = ssc.socketTextStream("localhost", 9999)
words = lines.flatMap(lambda line: line.split(" "))
word_counts = words.map(lambda word: (word, 1))

# 使用窗口函数，窗口大小为3秒，滑动间隔为1秒
windowed_word_counts = word_counts.reduceByKeyAndWindow(lambda a, b: a + b, lambda a, b: a - b, 3, 1)

windowed_word_counts.pprint()

ssc.start()
ssc.awaitTermination()

在这个示例中：

  • reduceByKeyAndWindow 方法用于在窗口上进行聚合操作。
  • 第一个参数是用于合并窗口内元素的函数，第二个参数是用于移除窗口外元素的函数。
  1. 状态更新
from pyspark import SparkContext
from pyspark.streaming import StreamingContext

sc = SparkContext("local[2]", "StatefulWordCount")
ssc = StreamingContext(sc, 1)
ssc.checkpoint("checkpoint")  # 启用检查点

def updateFunction(new_values, running_count):
    if running_count is None:
        running_count = 0
    return sum(new_values, running_count)

lines = ssc.socketTextStream("localhost", 9999)
words = lines.flatMap(lambda line: line.split(" "))
word_counts = words.map(lambda word: (word, 1))

# 使用updateStateByKey进行状态更新
stateful_word_counts = word_counts.updateStateByKey(updateFunction)

stateful_word_counts.pprint()

ssc.start()
ssc.awaitTermination()

在上述代码中：

  • updateStateByKey 方法用于维护每个键的状态。
  • updateFunction 定义了如何根据新值和现有状态更新状态。
  1. 与Kafka集成
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

sc = SparkContext("local[2]", "KafkaWordCount")
ssc = StreamingContext(sc, 1)

# Kafka参数
kafkaParams = {"metadata.broker.list": "localhost:9092"}
topics = ["test"]

# 创建Kafka输入DStream
kvs = KafkaUtils.createDirectStream(ssc, topics, kafkaParams)
lines = kvs.map(lambda x: x[1])

words = lines.flatMap(lambda line: line.split(" "))
word_counts = words.map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)

word_counts.pprint()

ssc.start()
ssc.awaitTermination()

在这个示例中：

  • KafkaUtils.createDirectStream 用于从Kafka主题读取数据。
  • kvs 是一个包含Kafka消息的DStream， lines 提取消息内容。

相关文章