Spark2.4.5 Structured Streaming消费Kafka不写入MySQL8.0及多SQL优化问题

针对Spark 2.4.5 + MySQL 8.0.20的环境，可按以下优先级排查：

• 依赖兼容性检查：确认pom.xml中引入的mysql-connector-java版本为8.0.x系列，不要使用5.x版本；集群提交作业时需保证驱动包在作业的classpath中，可在提交命令追加--jars mysql-connector-java-8.0.20.jar参数。
• JDBC配置检查：MySQL 8.0+需使用新驱动类和带时区的连接URL：
  驱动类固定为com.mysql.cj.jdbc.Driver
  连接URL格式参考：jdbc:mysql://<数据库地址>:<端口>/<库名>?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true
• 写入逻辑检查：
  若使用foreachBatch写法，确认微批DataFrame的写入逻辑调用了action算子（如write.jdbc），仅写转换逻辑不会触发执行；如果使用自定义ForeachWriter，需确认连接池配置合理、事务正确提交，不要为每条数据新建数据库连接。
  清空旧的checkpoint目录后重试，避免异常状态导致的写入停滞。
• 权限校验：确认Spark集群节点IP已加入MySQL访问白名单，作业运行账号拥有对应库表的写入权限。

不需要为每条SQL单独编写foreachBatch，可在同一个foreachBatch函数内复用当前微批的DataFrame完成多逻辑处理，示例参考如下：

streamingDF.writeStream
  .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    // 缓存微批数据，避免多次计算重复消费Kafka
    batchDF.persist()
    
    // 第一条SQL对应写入逻辑：写入明细表
    batchDF.select("id", "content", "create_time")
      .write.mode("append")
      .jdbc(jdbcUrl, "detail_table", jdbcProps)
    
    // 第二条SQL对应写入逻辑：聚合后写入统计报表
    batchDF.groupBy("create_date", "type").agg(count("id").alias("cnt"))
      .write.mode("append")
      .jdbc(jdbcUrl, "stats_table", jdbcProps)
    
    // 可继续追加更多写入逻辑
    // ...
    
    // 释放缓存
    batchDF.unpersist()
  }
  .option("checkpointLocation", "/your/checkpoint/path")
  .trigger(Trigger.ProcessingTime("1 minute")) // 按需配置触发间隔
  .start()
  .awaitTermination()

额外优化建议：

• JDBC写入参数新增batchsize配置，设置为1000~5000，开启批量写入提升吞吐量。
• 若不同写入逻辑的延迟要求、失败重试策略差异较大，可拆分多个独立的流作业处理，避免慢逻辑拖垮整个任务。
• 所有数据清洗、转换逻辑尽量放在foreachBatch外完成，减少微批内的计算耗时，避免作业背压。

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