如何通过Java Spark Pipeline写入BigQuery的JSON类型列？

可以实现，核心是使用新版Spark BigQuery Connector，并通过配置指定列类型映射，步骤如下：

1. 确保使用兼容的Connector版本 #

需要使用spark-bigquery-connector 0.28.0及以上版本（该版本开始支持BigQuery JSON类型的写入）。如果是Maven项目，添加依赖：

<dependency>
    <groupId>com.google.cloud.spark</groupId>
    <artifactId>spark-bigquery-with-dependencies_2.12</artifactId>
    <version>0.30.0</version>
</dependency>

2. 构造符合要求的DataFrame #

将需要写入JSON列的数据处理为合法的JSON字符串，存入Spark的StringType列——不要用StructType（会映射为RECORD）或直接用to_json后写入（会默认存为STRING）。

3. 写入时指定列类型映射 #

写入BigQuery时，通过配置明确指定目标列的类型为JSON，有两种可行方式：

方式一：显式指定表架构 #

如果目标表已存在或需要创建表并定义架构，使用setTableSchema参数强制映射列类型：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.util.Arrays;

public class BigQueryJsonWriter {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("Write BigQuery JSON Column")
                .getOrCreate();

        // 构造测试数据：JSON_COLUMN为合法JSON字符串
        Dataset<Row> df = spark.createDataFrame(
                Arrays.asList(
                        RowFactory.create(1, "{\"name\":\"Alice\",\"age\":30}"),
                        RowFactory.create(2, "{\"name\":\"Bob\",\"hobbies\":[\"reading\",\"hiking\"]}")
                ),
                new StructType(new StructField[]{
                        DataTypes.createStructField("ID", DataTypes.IntegerType, false),
                        DataTypes.createStructField("JSON_COLUMN", DataTypes.StringType, false)
                })
        );

        // 写入BigQuery，指定表架构将JSON_COLUMN映射为JSON类型
        df.write()
                .format("bigquery")
                .option("table", "your-project-id.your-dataset-id.target-table")
                .option("setTableSchema", "{\"fields\":[{\"name\":\"ID\",\"type\":\"INTEGER\",\"mode\":\"REQUIRED\"},{\"name\":\"JSON_COLUMN\",\"type\":\"JSON\",\"mode\":\"REQUIRED\"}]}")
                .mode("append")
                .save();
    }
}

方式二：启用自动JSON类型映射 #

如果你的Spark DataFrame中String列的内容均为合法JSON，可以通过Spark配置开启自动映射：

SparkSession spark = SparkSession.builder()
        .appName("Write BigQuery JSON Column")
        .config("spark.sql.bigquery.enableJsonType", "true")
        .getOrCreate();

// 后续写入无需额外指定schema，Connector会自动将内容为JSON的String列映射为BigQuery JSON类型
df.write()
        .format("bigquery")
        .option("table", "your-project-id.your-dataset-id.target-table")
        .mode("append")
        .save();

关键说明 #

• Spark的StructType会默认映射为BigQuery的RECORD类型，这与JSON类型是完全不同的结构，不能混用。
• to_json生成的String列默认会被写入为BigQuery的STRING类型，必须通过配置告知Connector将其识别为JSON类型才能正确写入目标列。

内容的提问来源于stack exchange，提问作者B-Brennan