PySpark整合Spark Streaming与Kafka：解决两大核心问题

针对你遇到的两个问题，以下是具体的代码调整方案：

问题1：从每个分区的最后提交偏移量开始读取 #

原代码未指定起始偏移量策略，默认会从最早偏移量（earliest）开始读取。要从消费者组已提交的最后偏移量启动，需添加startingOffsets配置，并配合Spark checkpoint机制确保偏移量持久化：

修改后的代码片段 #

# 创建SparkSession（保持原代码不变）
spark = SparkSession.builder \
    .appName(appName) \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1") \
    .getOrCreate()

# 定义Schema（保持原代码不变）
schema = StructType([
            StructField("col1", StringType()),
            StructField("col2", StringType()),
            StructField("col3", TimestampType()), 
            StructField("col4", DoubleType())
        ])

# 从消费者组已提交偏移量开始读取Kafka流
df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", broker) \
    .option("subscribe", topic) \
    .option("kafka.group.id", appName) \
    .option("startingOffsets", "group-offsets")  # 关键：使用消费者组已提交的偏移量
    .option("enable.auto.commit", True) \
    .load()

value_df = df.select(col("topic"), col("partition"), col("offset"), col("timestamp"),
                     from_json(col("value").cast("STRING"), schema).alias("values"))

# 启用Spark Checkpoint持久化偏移量（推荐，避免依赖Kafka自动提交的不可靠性）
query = value_df.writeStream \
    .outputMode("append") \
    .option("checkpointLocation", "/your/checkpoint/path")  # 替换为实际路径（HDFS/本地磁盘）
    .format("console")  # 可替换为你需要的输出格式（如parquet、jdbc等）
    .start()

query.awaitTermination()

关键说明 #

• startingOffsets="group-offsets"：让Spark使用指定消费者组（kafka.group.id）在Kafka中已提交的偏移量作为起始点
• Checkpoint机制：将偏移量持久化到指定路径，即使重启应用也能从断点继续读取，比Kafka自动提交更可靠

问题2：读取最近15分钟的流数据 #

有两种实现方式，推荐优先使用Kafka端过滤（减少数据传输量）：

方式1：Kafka端按时间戳过滤（推荐） #

通过计算15分钟前的时间戳，让Kafka只返回该时间点之后的消息：

from datetime import datetime, timedelta

# 计算15分钟前的毫秒级时间戳
start_timestamp_ms = int((datetime.now() - timedelta(minutes=15)).timestamp() * 1000)

df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", broker) \
    .option("subscribe", topic) \
    .option("startingTimestamp", str(start_timestamp_ms))  # 从15分钟前开始读取
    .option("startingOffsets", "latest")  # 兜底：如果指定时间戳无对应偏移量，从最新偏移量开始
    .load()

方式2：Spark端过滤消息时间 #

读取所有消息后，过滤出最近15分钟内的数据（适用于需要同时保留历史数据校验的场景）：

from pyspark.sql.functions import current_timestamp, expr

# 解析消息后过滤最近15分钟的数据
filtered_df = value_df \
    .select("*", "values.*") \
    # 用Kafka消息自带的timestamp过滤，若业务时间是col3，替换为col("col3")
    .where(col("timestamp") >= current_timestamp() - expr("INTERVAL 15 MINUTES"))

内容的提问来源于stack exchange，提问作者Kavya shree