4、数据接入层设计：使用Spark Streaming或Structured Streaming读取Kafka行情Topic，实现Exactly-Once语义

高频行情数据接入层是整个标签计算管线的第一道关卡。行情数据以毫秒级速率写入Kafka Topic，接入层必须同时满足高吞吐、低延迟和精确一次（Exactly-Once）的语义要求。本节将详细阐述如何基于Spark Structured Streaming（推荐）或Spark Streaming（传统方案）对接Kafka，并保证数据不重不丢。

4.1 接入层架构总览

数据接入层位于Kafka与标签计算引擎之间，核心职责包括：

• 订阅行情Topic：按分区并行拉取实时行情消息。
• 反序列化与Schema校验：将二进制消息转换为DataFrame行。
• 水位线与时间戳处理：为后续窗口聚合提供事件时间语义。
• 偏移量管理：通过Checkpoint或事务性写入实现Exactly-Once。

4.2 Exactly-Once 语义的技术原理

在流处理系统中，Exactly-Once 意味着每条消息恰好被处理一次，不会因故障导致重复或遗漏。Spark Structured Streaming 通过以下机制实现该语义：

4.3 使用 Structured Streaming 读取 Kafka

以下代码展示了如何配置 SparkSession 并创建读取 Kafka 行情 Topic 的流式 DataFrame：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, TimestampType, DoubleType

# 1. 创建 SparkSession，开启 Checkpoint
spark = SparkSession.builder \
    .appName("TickDataIngestion") \
    .config("spark.sql.streaming.checkpointLocation", "hdfs://namenode:8020/checkpoint/tick_ingest") \
    .config("spark.sql.streaming.schemaInference", "false") \
    .getOrCreate()

# 2. 定义行情消息 Schema（示例）
tick_schema = StructType([
    StructField("symbol", StringType(), True),
    StructField("timestamp", TimestampType(), True),
    StructField("price", DoubleType(), True),
    StructField("volume", DoubleType(), True),
    StructField("side", StringType(), True)   # 'B' 买 / 'S' 卖
])

# 3. 从 Kafka 读取流数据
df_raw = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092") \
    .option("subscribe", "tick_data") \
    .option("startingOffsets", "latest") \
    .option("failOnDataLoss", "false") \
    .load()

# 4. 解析二进制 value 为结构化数据
from pyspark.sql.functions import from_json, col

df_parsed = df_raw \
    .select(from_json(col("value").cast("string"), tick_schema).alias("data")) \
    .select("data.*")

# 5. 添加 Kafka 元数据（分区、偏移量）用于审计
df_with_meta = df_parsed \
    .withColumn("kafka_partition", col("partition")) \
    .withColumn("kafka_offset", col("offset"))

4.4 关键配置参数说明

接入层性能与可靠性高度依赖以下参数的合理设置：

4.5 写入下游存储的 Exactly-Once 策略

接入层解析后的数据需要写入标签计算引擎（如 Delta Lake 或内存表）。写入端也必须支持幂等或事务性语义：

• Delta Lake 写入：使用 foreachBatch + DeltaTable.alias 的 merge 操作，按主键去重。
• Kafka 回写：启用 kafka.enable.idempotence=true，配合 acks=all。
• HDFS 文件写入：使用 FileStreamSink 配合 _SUCCESS 标记文件，避免部分写入。

# 示例：使用 foreachBatch 写入 Delta Lake（幂等合并）
def write_to_delta(df, epoch_id):
    df.write \
      .format("delta") \
      .mode("append") \
      .option("mergeSchema", "true") \
      .save("/data/delta/tick_raw")

query = df_with_meta.writeStream \
    .foreachBatch(write_to_delta) \
    .outputMode("append") \
    .trigger(processingTime="5 seconds") \
    .start()

query.awaitTermination()

4.6 故障恢复与 Exactly-Once 验证

当作业因网络、资源或代码异常重启时，恢复流程如下：

1. Spark 从 Checkpoint 目录读取上次成功提交的批次 ID 和偏移量范围。
2. 重新创建 Kafka 消费者，从记录的偏移量开始拉取。
3. 重复执行未完成的批次，但写入操作通过幂等性保证不会产生重复数据。
4. 如果使用 Delta Lake，事务日志会跳过已提交的文件，实现精确一次。

4.7 性能调优建议

• 增加 Kafka 分区数：分区数应 ≥ Spark 执行器核心数，避免分区成为瓶颈。
• 调整微批次间隔：高频行情建议 trigger(processingTime="2-5 seconds")，平衡延迟与吞吐。
• 禁用不必要的 Schema 推断：显式指定 Schema，避免 spark.sql.streaming.schemaInference=true 带来的性能开销。
• 使用 Kyro 序列化：配置 spark.serializer=org.apache.spark.serializer.KryoSerializer 减少网络传输开销。