Spark Structured Streaming读Kafka数据遇OOM，求限流读取方案

使用Spark Structured Streaming从Kafka主题读取约1000万条数据，转换后写入MySQL时出现java.lang.OutOfMemoryError: GC overhead limit exceeded错误，希望限制单次读取的记录数避免内存溢出。相关代码如下：

PySpark配置 #

conf = SparkConf().setAppName("KafkaStreamingDataPipeline")
        .setMaster("local[*]") 
        .set("spark.driver.cores","8")
        .set("spark.driver.maxResultSize","8g")
        .set("spark.driver.memory","15g")
        .set("spark.executor.memory","8g")
        .set("spark.executor.instances", "8")
        .set("spark.executor.cores", "8")

spark = SparkSession.builder.master("local[*]").appName("Kafka Pyspark Streaming Learning").config(conf=conf).getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

读流代码 #

orders_df = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
        .option("subscribe", KAFKA_TOPIC_NAME) \
        .option("startingOffsets", "earliest") \
        .option("failOnDataLoss", "false") \
        .load()

写流代码 #

trans_detail_write_stream = dfTrans2.writeStream \
        .trigger(processingTime='25 seconds') \
        .outputMode("append") \
        .foreachBatch(saveToMysql) \
        .start()

saveToMysql函数 #

def saveToMysql(currentDf, epocId):

    dbCredentials = {"user": mysql_user_name,
                     "password":mysql_password,
                     "driver":mysql_driver_class}

    print("Printing Epoc Id")
    print(epocId)

    processedAt = time.strftime("%Y-%m-%d %H:%M:%S")

    currentDfFinal = currentDf\
        .withColumn("processed_at",lit(processedAt))\
        .withColumn("batch_id", lit(epocId))

    print("Printing Before MySql Table Save: " + str(epocId))

    currentDfFinal \
        .write \
        .jdbc(url=mysqlJdbcUrl,
              table = mysql_table_name,
              mode = "append",
              properties = dbCredentials)

    print("Printing After MySql table Save " + str(epocId))

要限制Structured Streaming从Kafka单次读取的数据量，可通过以下几种方式实现：

1. 配置Kafka消费者拉取上限 #

在Kafka读流参数中添加kafka.max.poll.records，控制消费者单次从Kafka拉取的最大记录数，避免一次性加载过多数据到内存：

orders_df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
    .option("subscribe", KAFKA_TOPIC_NAME) \
    .option("startingOffsets", "earliest") \
    .option("failOnDataLoss", "false") \
    .option("kafka.max.poll.records", "100000")  # 单次拉取10万条，可根据内存情况调整
    .load()

2. 限制单分区处理速率 #

通过全局配置spark.streaming.kafka.maxRatePerPartition，控制每个Kafka分区每秒处理的最大记录数，结合触发器的处理时间，间接控制每个批次的数据量：

# 先设置全局参数
spark.conf.set("spark.streaming.kafka.maxRatePerPartition", "50000")  # 每个分区每秒最多处理5万条
spark.conf.set("spark.sql.shuffle.partitions", "32")  # 调整shuffle分区数，降低内存压力

# 写流可配合缩短处理间隔，拆分批次
trans_detail_write_stream = dfTrans2.writeStream \
    .trigger(processingTime='10 seconds')
    .outputMode("append") \
    .foreachBatch(saveToMysql) \
    .start()

3. 在foreachBatch内部拆分写入 #

如果批次数据量仍过大，可在saveToMysql函数内将DataFrame拆分为多个小批次写入MySQL，避免一次性写入大量数据：

def saveToMysql(currentDf, epocId):
    dbCredentials = {"user": mysql_user_name,
                     "password":mysql_password,
                     "driver":mysql_driver_class}

    processedAt = time.strftime("%Y-%m-%d %H:%M:%S")
    currentDfFinal = currentDf\
        .withColumn("processed_at",lit(processedAt))\
        .withColumn("batch_id", lit(epocId))
    
    # 拆分DataFrame为小批次写入
    batch_size = 50000
    # 用randomSplit拆分，比offset+limit效率更高
    splits = currentDfFinal.randomSplit([1.0]*((currentDfFinal.count()//batch_size)+1))
    
    for idx, batch_df in enumerate(splits):
        batch_df.write \
            .jdbc(url=mysqlJdbcUrl,
                  table=mysql_table_name,
                  mode="append",
                  properties=dbCredentials)
        print(f"完成子批次 {idx+1} 写入，epochId: {epocId}")

4. 辅助内存优化 #

• 避免在生产场景使用local[*]模式，本地模式下driver与executor共用进程，内存压力集中，建议提交到集群运行。
• 检查spark.driver.memory与spark.executor.memory的分配，确保executor有足够内存处理数据，同时避免driver内存过载。

内容的提问来源于stack exchange，提问作者Muhammad Affan