使用Apache Beam（Scio）导入含RECORD类型BigQuery表的报错及可行性咨询

用Apache Beam（包括Scio）导入含RECORD（STRUCT）类型的BigQuery表完全可行，不存在固有限制，你遇到的问题是Scio导出JSON时对STRUCT字段的序列化方式错误导致的。

问题根源 #

你看到的扁平字符串格式（例如"organization_fields":"{org_status_acordo=false, org_inadimp=false, org_status_averb=false, cnpj_do_cadastro=}"），是Scio默认将对应STRUCT的Scala case class序列化成了toString()的结果，而非标准嵌套JSON结构。BigQuery导入时要求RECORD字段必须是嵌套JSON对象，因此触发报错Flat value specified for record field。

解决方法 #

要让Scio导出符合要求的嵌套JSON，可通过以下两种方式调整：

• 自定义JSON序列化逻辑：如果必须生成中间JSON文件，替换默认序列化方式，用Jackson或Circe等库将case class序列化为标准嵌套JSON。示例（Jackson实现）：

  import com.fasterxml.jackson.databind.ObjectMapper
  import com.fasterxml.jackson.module.scala.DefaultScalaModule
  import com.spotify.scio.values.SCollection
  
  // 初始化支持Scala类型的Jackson ObjectMapper
  val jsonMapper = new ObjectMapper().registerModule(DefaultScalaModule)
  // 将数据集中的对象序列化为标准JSON字符串
  val standardJson: SCollection[String] = yourDataset.map(jsonMapper.writeValueAsString)
  // 导出到存储
  standardJson.saveAsTextFile("gs://your-bucket/target-path")
  

• 直接写入BigQuery：跳过中间JSON环节，用Scio的BigQuery原生API直接将case class映射到BigQuery表结构，避免序列化格式问题。示例：

  case class OrganizationFields(
    org_status_acordo: Boolean,
    org_inadimp: Boolean,
    org_status_averb: Boolean,
    cnpj_do_cadastro: Option[String]
  )
  case class BusinessData(id: String, organization_fields: OrganizationFields)
  
  sc.parallelize(yourDataList)
    .saveAsBigQuery(
      tableSpec = "your-project:your-dataset.target-table",
      schema = BigQuerySchema.fromCaseClass[BusinessData],
      writeDisposition = WriteDisposition.WRITE_TRUNCATE
    )
  

关键提示 #

如果必须通过中间JSON文件导入BigQuery，务必确保RECORD对应的字段是嵌套JSON对象（格式示例："organization_fields":{"org_status_acordo":false,"org_inadimp":false,...}），而非字符串形式的扁平结构。

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