Spark写入Snowflake遇数据不匹配，求过滤无效记录的解决方案

1. 提前在Spark侧校验并过滤无效数据 #

在写入Snowflake之前，先对DataFrame的数据进行校验，过滤掉name字段长度超过8的记录，同时可记录无效数据用于后续排查：

import org.apache.spark.sql.functions._

// 给DataFrame指定与Snowflake表匹配的列名
val dfFromRDD1 = rdd.toDF("id", "name", "price")

// 筛选出name字段长度超过8的无效记录
val invalidRecords = dfFromRDD1.filter(length(col("name")) > 8)

// 输出无效记录信息（可选：也可写入文件/临时表留存）
if (invalidRecords.count() > 0) {
  println(s"检测到${invalidRecords.count()}条无效记录（name字段长度超过8）：")
  invalidRecords.show(false)
}

// 过滤得到符合要求的有效数据
val validDF = dfFromRDD1.filter(length(col("name")) <= 8)

// 将有效数据写入Snowflake
validDF.write.format("net.snowflake.spark.snowflake")
  .options(sfOptions)
  .option("dbtable", "product")
  .mode(SaveMode.Append)
  .save()

2. 利用Snowflake连接器参数自动处理数据不匹配问题 #

如果不想手动过滤，也可以通过连接器参数让Snowflake自动处理这类问题：

自动截断超长字符串 #

设置truncatecolumns选项为true，Snowflake会自动将超长的字符串截断到列定义的长度：

validDF.write.format("net.snowflake.spark.snowflake")
  .options(sfOptions)
  .option("dbtable", "product")
  .option("truncatecolumns", "true")
  .mode(SaveMode.Append)
  .save()

跳过错误记录 #

通过copy_options设置ON_ERROR = 'CONTINUE'，让Snowflake跳过写入失败的记录（错误信息会留存于Snowflake的加载历史中）：

validDF.write.format("net.snowflake.spark.snowflake")
  .options(sfOptions)
  .option("dbtable", "product")
  .option("copy_options", "ON_ERROR = 'CONTINUE'")
  .mode(SaveMode.Append)
  .save()

3. 优化错误提示清晰度 #

默认情况下连接器可能不会返回详细错误，开启Snowflake连接器的DEBUG日志可以直接在Spark日志中看到具体错误原因：

// 在代码中设置日志级别
spark.sparkContext.setLogLevel("DEBUG")
// 或者在提交任务时添加参数
// spark-submit --conf log4j.logger.net.snowflake=DEBUG ...

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