Spark生成Avro导入BigQuery遇long转int转换错误求解决方案

首先，先解答你的疑问：为什么报错里是"long"和"int"，而不是和BigQuery的"INTEGER"？

这个错误是Apache Avro解析库直接抛出的，它用的是Avro自身的类型术语（int是32位整数，long是64位整数），而不是BigQuery的类型名称。BigQuery在加载Avro数据时会调用Avro库解析数据，当Avro库发现实际数据的类型和期望的类型（不管是自动推断的还是你指定的）不兼容时，就会抛出这个错误，所以错误信息里用的是Avro的类型名词，而非BigQuery的INTEGER。

接下来分析问题原因和解决办法：

可能的原因 #

1. Spark写Avro时的类型映射偏差：Spark的IntegerType对应Avro的int（32位），LongType对应Avro的long（64位）。如果你的Spark DataFrame里C6-C8是IntegerType，但数据实际值超出了32位范围，或者Spark Avro序列化时生成了nullable的union类型，就可能导致Avro文件里的字段类型和你预期的不一致。
2. BigQuery自动检测schema的推断错误：当你用--autodetect时，BigQuery会扫描样本数据推断类型，如果样本里的数值都是小整数，它可能推断成Avro的int，但实际Avro文件里存在long类型的数值，导致类型不匹配。
3. 字段名大小写不匹配：Avro字段名是大小写敏感的，如果Spark写的Avro字段是小写（比如c6），但你指定BigQuery schema时用了大写（C6），会导致BigQuery找不到对应字段，进而触发类型解析错误。

解决步骤 #

第一步：确认Avro文件的实际schema #

先搞清楚你的Avro文件里每个字段的真实类型，这是排查的关键：

• 用avro-tools命令查看（需要先下载avro-tools工具）：

  avro-tools getschema gs://test/avrodebug/your-sample-file.avro
  

• 或者在Spark里读取Avro文件打印schema：

  val df = spark.read.avro("gs://test/avrodebug/*.avro")
  df.printSchema()
  

第二步：针对性解决问题 #

根据Avro schema的结果，选择对应的方案：

方案1：调整Spark写Avro的类型

如果发现Avro里C6-C8是int，但数据可能超出32位范围，建议在Spark里把这些字段改成LongType，这样Avro里会生成long类型，和BigQuery的INTEGER完美兼容：

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

// 转换DataFrame字段类型
val adjustedDf = dataframe
  .withColumn("C6", col("C6").cast(LongType))
  .withColumn("C7", col("C7").cast(LongType))
  .withColumn("C8", col("C8").cast(LongType))

// 写入Avro
adjustedDf.write.avro("path")

方案2：精确指定BigQuery schema（避免自动推断）

不要用--autodetect，而是根据Avro的schema精确指定BigQuery的schema，推荐用JSON格式的schema文件，避免逗号分隔格式的潜在问题：

1. 创建schema.json文件：

[
  {"name": "C1", "type": "STRING"},
  {"name": "C2", "type": "STRING"},
  {"name": "C3", "type": "STRING"},
  {"name": "C4", "type": "STRING"},
  {"name": "C5", "type": "STRING"},
  {"name": "C6", "type": "INTEGER"},
  {"name": "C7", "type": "INTEGER"},
  {"name": "C8", "type": "INTEGER"},
  {"name": "C9", "type": "STRING"},
  {"name": "C10", "type": "STRING"},
  {"name": "C11", "type": "STRING"}
]

2. 执行加载命令：

bq --nosync load --source_format AVRO datasettest.testtable gs://test/avrodebug/*.avro schema.json

方案3：处理nullable的union类型

如果Avro schema里的字段是union类型（比如["null", "long"]），可以在Spark写Avro时强制指定schema，避免自动生成复杂的union类型：

import org.apache.avro.SchemaBuilder

// 手动定义Avro schema，确保字段类型明确
val avroSchema = SchemaBuilder.record("record").fields()
  .name("C1").type().stringType().noDefault()
  .name("C2").type().stringType().noDefault()
  .name("C3").type().stringType().noDefault()
  .name("C4").type().stringType().noDefault()
  .name("C5").type().stringType().noDefault()
  .name("C6").type().longType().noDefault()
  .name("C7").type().longType().noDefault()
  .name("C8").type().longType().noDefault()
  .name("C9").type().stringType().noDefault()
  .name("C10").type().stringType().noDefault()
  .name("C11").type().stringType().noDefault()
  .endRecord()

// 写入Avro时指定schema
dataframe.write
  .format("avro")
  .option("avroSchema", avroSchema.toString)
  .save("path")

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