基于Spark Structured Streaming的股票价格告警应用技术问题咨询

项目概述 #

我正在开展大数据课程项目，开发基于Spark Streaming和Kafka的股票价格邮件告警应用。

股票交易数据示例 #

stockName,tradeType,price,quantity,timestamp 
MQG,SELL,940.22,25,2023-11-06 00:01:00.000

Kafka生产者代码 #

def send_data():
    producer = Producer({'bootstrap.servers': 'localhost:9092'})
    topic = "stream-price"
    df = pd.read_csv('../../data/stock_trading_data.csv')

    for index, row in df.iterrows():
        dict_stock = json.dumps(row.to_dict()).encode('utf-8')
        try:
            producer.produce(topic, value=dict_stock)
        except Exception as e:
            print(f"Error producing message: {e}")

        sleep_time = 10  # each 30 seconds
        sleep(sleep_time)

    producer.flush()
    producer.close()

Spark结构化流读取逻辑 #

股票数据流

data_stream = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", topic_streaming) \
    .option("startingOffsets", "earliest") \
    .load()
data_json_schema = StructType([
    StructField("stockName", StringType(), True),
    StructField("tradeType", StringType(), True),
    StructField("price", DoubleType(), True),
    StructField("quantity", LongType(), True),
    StructField("timestamp", TimestampType(), True),
])

data_stream = data_stream.selectExpr("cast(value as string) as value")
data_stream = data_stream.withColumn("value", from_json(
    data_stream["value"], data_json_schema)).select("value.*")
data_stream = data_stream.withColumn("timestamp", to_timestamp(
    data_stream["timestamp"], "HH:mm:ss"))

用户配置流

require_stream = spark \
    .readStream \
    .format("socket") \
    .option("host", "localhost") \
    .option("port", 9090) \
    .load()
require_json_schema = StructType([
    StructField("stockNameRule", StringType()),
    StructField("tradeTypeRule", StringType(), True),
    StructField("priceRule", DoubleType()),
    StructField("quantityRule", LongType(), True),
])

require_stream = require_stream \
    .select(from_json("value", configs_json_schema).alias("data")) \
    .select("data.*")

告警逻辑设计 #

计划通过流-流关联+过滤实现告警，代码如下：

join_stream = data_stream.join(
    configs_stream,
    expr("""
         stockName = stockNameRule AND
         price >= priceRule
         """)
join_stream = df_with_stateful \
     .writeStream \
     .foreach(process_row) \ # send email func
     .start()

核心问题解决方案 #

1. 避免重复发送告警，仅针对最新配置下的匹配数据发送单封邮件 #

核心思路是用状态管理跟踪每个股票+配置组合的告警历史，确保只有符合最新配置的新数据触发告警，具体实现：

• 第一步：过滤出最新用户配置
  用户可能更新配置，需先对配置流去重，保留每个股票规则的最新版本：

from pyspark.sql.functions import current_timestamp, max, first

# 给配置流添加接收时间戳，用于判断最新配置
require_stream = require_stream.withColumn("config_receive_time", current_timestamp())

# 分组保留每个股票规则的最新配置
latest_configs = require_stream.groupBy("stockNameRule")\
    .agg(
        max("config_receive_time").alias("latest_config_time"),
        first("tradeTypeRule").alias("tradeTypeRule"),
        first("priceRule").alias("priceRule"),
        first("quantityRule").alias("quantityRule")
    )

• 第二步：用状态管理控制告警触发
  使用flatMapGroupsWithState维护告警状态，仅当新数据满足最新配置且未触发过告警时发送邮件：

from pyspark.sql.streaming import GroupState, GroupStateTimeout

# 定义状态类，存储告警历史和配置版本
class AlertState:
    def __init__(self, last_alert_time, latest_config_time):
        self.last_alert_time = last_alert_time
        self.latest_config_time = latest_config_time

def update_alert_state(key, values, state: GroupState[AlertState]):
    # key为(stockName, stockNameRule)，唯一标识股票+配置组合
    latest_data = max(values, key=lambda x: x.timestamp)
    current_config_time = latest_data.latest_config_time

    if state.exists:
        current_state = state.get()
        # 仅当配置更新或数据晚于上次告警时触发
        if current_config_time > current_state.latest_config_time or \
            latest_data.timestamp > current_state.last_alert_time:
            process_row(latest_data)
            state.update(AlertState(latest_data.timestamp, current_config_time))
            return [(key[0], latest_data.price, "alert_sent")]
    else:
        # 首次匹配，直接发送并初始化状态
        process_row(latest_data)
        state.set(AlertState(latest_data.timestamp, current_config_time))
        return [(key[0], latest_data.price, "alert_sent")]
    return []

# 关联股票流和最新配置流，过滤符合条件的数据
joined_stream = data_stream.join(
    latest_configs,
    data_stream.stockName == latest_configs.stockNameRule
).filter(
    (data_stream.price >= latest_configs.priceRule) &
    (data_stream.tradeType == latest_configs.tradeTypeRule)
)

# 按股票+配置分组，应用状态更新逻辑
alert_stream = joined_stream.groupBy("stockName", "stockNameRule")\
    .flatMapGroupsWithState(GroupStateTimeout.NoTimeout, update_alert_state)

# 启动流任务
query = alert_stream.writeStream.outputMode("append").format("console").start()
query.awaitTermination()

• 状态清理优化
  给股票数据流添加水印，自动清理过期状态，避免内存溢出：

data_stream = data_stream.withWatermark("timestamp", "10 minutes")

2. 处理股票数据乱序问题 #

利用水印（Watermark）+ 事件时间窗口是处理乱序数据的标准方案，具体实现：

• 第一步：设置水印容忍时间
  指定乱序数据的最大延迟时间，Spark会自动忽略超过该时间的旧数据：

# 基于事件时间设置水印，容忍5分钟乱序
data_stream = data_stream.withWatermark("timestamp", "5 minutes")

• 第二步：窗口聚合处理乱序数据
  若需按时间维度统计（比如窗口最高价格告警），定义滚动/滑动窗口，水印会自动清理过期窗口状态：

from pyspark.sql.functions import window, max

# 定义1分钟滚动窗口，统计窗口内最高价格
windowed_data = data_stream.groupBy(
    "stockName",
    window("timestamp", "1 minute")
).agg(max("price").alias("window_max_price"))

# 关联最新配置流，过滤满足条件的窗口数据
window_alert_stream = windowed_data.join(
    latest_configs,
    windowed_data.stockName == latest_configs.stockNameRule
).filter(windowed_data.window_max_price >= latest_configs.priceRule)

# 发送告警
query = window_alert_stream.writeStream.foreach(process_row).start()
query.awaitTermination()

• 无聚合场景的乱序处理
  若无需聚合，直接处理单条乱序数据，可在状态管理中过滤水印外的旧数据：

def update_state(key, values, state):
    # 过滤超过水印容忍时间的旧数据
    watermark_time = current_timestamp() - expr("interval 5 minutes")
    valid_values = [v for v in values if v.timestamp >= watermark_time]
    if not valid_values:
        return []

    # 取当前批次最新有效数据
    latest_data = max(valid_values, key=lambda x: x.timestamp)
    # 后续告警判断逻辑同状态管理代码

内容的提问来源于stack exchange，提问作者Baubau Tran