如何使用Spark Scala写入BigQuery的JSON列及实现示例

核心原因 #

BigQuery的JSON列要求写入时的数据类型需与表定义匹配，报错是因为当前写入的DataFrame列类型为STRING，而目标表该列定义为JSON，连接器默认不会自动转换，需显式配置或转换。

方案1：通过连接器配置自动映射类型 #

在写入BigQuery时，添加spark.sql.bigquery.typeMapping.jsonAsString配置，让连接器将Spark的StringType列自动映射为BigQuery的JSON列。

示例代码：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("Write JSON to BigQuery")
  .getOrCreate()

// 假设你的DataFrame是df，包含名为source的StringType列（存储JSON字符串）
val df = spark.read... // 读取你的数据源

// 写入BigQuery的配置
df.write
  .format("bigquery")
  .option("table", "your-project.your-dataset.your-table")
  .option("writeMethod", "direct")
  .option("spark.sql.bigquery.typeMapping.jsonAsString", "true") // 关键配置
  .mode("append")
  .save()

方案2：显式指定目标表Schema（适用于首次创建表或Schema变更） #

如果目标表还未创建，或者需要确保Schema匹配，可以显式指定Schema，将列定义为JSON类型：

示例代码：

import org.apache.spark.sql.types.{StructType, StructField, StringType}

// 定义Schema，source列类型为StringType，后续通过配置映射为BigQuery JSON
val schema = new StructType()
  .add(StructField("id", StringType, nullable = false))
  .add(StructField("source", StringType, nullable = true))

val df = spark.read.schema(schema)... // 读取数据源

df.write
  .format("bigquery")
  .option("table", "your-project.your-dataset.your-table")
  .option("writeMethod", "direct")
  .option("spark.sql.bigquery.typeMapping.jsonAsString", "true")
  .option("schema", """{"fields":[{"name":"id","type":"STRING","mode":"REQUIRED"},{"name":"source","type":"JSON","mode":"NULLABLE"}]}""") // 显式指定BigQuery Schema
  .mode("append")
  .save()

方案3：手动转换为Spark的JSON类型（Spark 3.2+支持） #

Spark 3.2及以上版本支持JsonType，可以将String列转换为JsonType后再写入，此时连接器会自动映射为BigQuery的JSON列：

import org.apache.spark.sql.functions.from_json
import org.apache.spark.sql.types.JsonType

// 将source列从StringType转为JsonType
val transformedDf = df.withColumn("source", from_json(df("source"), JsonType))

transformedDf.write
  .format("bigquery")
  .option("table", "your-project.your-dataset.your-table")
  .option("writeMethod", "direct")
  .mode("append")
  .save()

注意事项 #

• 确保spark-bigquery-connector版本在0.20.0及以上（支持JSON列映射）
• 如果目标表已存在且列类型为JSON，必须保证写入的数据是合法的JSON字符串，否则会写入失败
• 追加模式下，需确保DataFrame的列类型与目标表列类型通过配置或转换匹配，避免类型冲突

内容的提问来源于stack exchange，提问作者sarath.mec