Spark读取HDFS CSV数据写入MySQL方案求验证及优化建议

整体流程的核心逻辑是可行的，符合HDFS->Spark处理->MySQL的链路要求，但存在1个逻辑错误，以及多处可优化的点，具体如下：

1. 首要修复的逻辑错误 #

你完成过滤、字段选择后的结果存在resultsdf变量中，但最终写入MySQL时调用的是原始读取的df，相当于数据处理逻辑完全没有生效，需要修改为：

// 原错误写法 df.write.mode(SaveMode.Append).jdbc(url,"sample_metrics",prop)
// 修正后
resultsdf.write.mode(SaveMode.Append).jdbc(url,"sample_metrics",prop)

2. 读取CSV环节优化 #

• 不要在生产环境使用inferSchema="true"：该参数会额外扫描全量数据做类型推断，拖慢读取速度，且推断的字段类型可能不符合预期（比如数值溢出、日期识别错误），建议手动指定Schema：

import org.apache.spark.sql.types._
val customSchema = StructType(Array(
  StructField("Sample", StringType, nullable = true),
  StructField("p16", DoubleType, nullable = true),
  StructField("Age", IntegerType, nullable = true),
  StructField("Race", StringType, nullable = true),
  StructField("Anatomy", StringType, nullable = true)
))
val df = spark.read.format("csv")
    .option("header", "true")
    .schema(customSchema) // 替换inferSchema
    .option("nullValue","NA")
    .option("mode","failfast")
    .option("badRecordsPath", "hdfs://localhost:9000/user/testuser/bad_records") // 脏数据落盘方便排查
    .load("hdfs://localhost:9000/user/testuser/samples.csv")

• 可根据业务容错需求调整读取模式：如果允许脏数据存在，可以将mode改为PERMISSIVE（错误字段设为null）或DROPMALFORMED（直接丢弃脏行）。

3. 写入MySQL环节优化 #

• 敏感信息安全优化：不要在代码里硬编码数据库账号密码，建议通过Spark提交参数传入，比如提交时加--conf spark.jdbc.user=xxx --conf spark.jdbc.password=xxx，代码中通过spark.conf.get("spark.jdbc.user")获取，避免密码泄露。
• 写入性能优化：
  • JDBC URL添加批量提交参数，开启批量写入：val url = "jdbc:mysql://localhost:3306/meta?rewriteBatchedStatements=true&useUnicode=true&characterEncoding=utf8"，后面的编码参数还可以避免中文乱码。
  • 调整批量写入大小：给Properties添加prop.setProperty("batchsize", "2000")，根据单行数据大小可调整为1000~10000区间。
  • 控制并行写入连接数：如果数据量较大，写入前对结果做重分区，避免并行度过高打爆MySQL连接池，比如resultsdf.repartition(8).write.xxx，分区数不要超过MySQL实例的最大连接数限制。
• 写入逻辑优化：
  • 当前Append模式会导致任务重复执行时产生重复数据，如果是全量更新场景可以改为SaveMode.Overwrite；如果是增量更新场景，建议给MySQL表加唯一键约束，或者写入前先清理对应增量范围的数据。

4. 可观测性优化 #

• 代码中增加数据量校验日志，比如打印df.count()、resultsdf.count()，方便核对数据量是否符合预期。
• 增加异常捕获逻辑，写入失败时触发告警，避免任务失败影响报表产出。

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