版本：Next

BigQuery

BigQuery 数据接收器连接器

支持的引擎

Spark
Flink
Seatunnel Zeta

主要特性

描述

用于 Google Cloud BigQuery 的数据接收器连接器,使用 Storage Write API 实现高性能数据摄取。

支持的数据源信息

数据源	支持的版本	Maven
BigQuery	BOM 26.72.0	下载

配置选项

名称	类型	是否必须	默认值	描述
project_id	string	是	-	GCP 项目 ID
dataset_id	string	是	-	BigQuery 数据集 ID
table_id	string	是	-	BigQuery 表 ID
service_account_key_path	string	否	-	GCP 服务账号 JSON 密钥文件路径
service_account_key_json	string	否	-	内联 GCP 服务账号 JSON 密钥内容
write_mode	string	否	batch	写入模式。支持的值：`batch` 和 `streaming`
sequence_number_column	string	否	-	用于 CDC 去重的序列号列名。仅在 `write_mode` 为 `streaming` 时适用
batch_size	int	否	1000	发送到 BigQuery 之前批量处理的行数
emulator_host	string	否	-	BigQuery emulator 地址，例如 `localhost:9050`。该参数仅用于测试。
multi_table_sink_replica	int	否	-	Sink 通用参数，用于控制多表运行时每张表的 sink 副本数；但该连接器仍只写入配置中的单个 BigQuery 表。
common-options		否	-	Sink 通用参数，详见 Sink Common Options。

认证参数

生产 BigQuery 任务必须使用下面任意一种认证方式。只有配置 emulator_host 做测试时才会跳过认证。

service_account_key_path：服务账号 JSON 密钥文件路径。
service_account_key_json：直接填写服务账号 JSON 密钥内容。
默认凭据：如果前两项都不配置，则使用 Google Application Default Credentials。

表选项

目标 BigQuery 表必须已经存在。连接器会在 writer 初始化时读取已有的表 schema，并且不会自动创建 BigQuery 表。

该连接器会写入一个固定的目标表：project_id.dataset_id.table_id。它不会按上游表自动创建或切换 BigQuery 目标表。如果任务里有多张表，请配置多个 BigQuery sink，或者在写入 BigQuery 前先完成表路由。

写入模式

batch：使用 BigQuery buffered write stream，并在 SeaTunnel checkpoint/commit 阶段提交数据。主要特性中的精确一次能力指的是该模式。
streaming：使用默认 stream，并携带 BigQuery change 字段写入 CDC 记录。该模式适合 CDC 的 upsert/delete 数据，但该连接器没有将它标记为精确一次。

使用 streaming 模式写入 CDC 数据时，请先在 BigQuery 中创建好带 Primary Key 的目标表。连接器会把 SeaTunnel 的行类型转换为 BigQuery change 记录：INSERT 和 UPDATE_AFTER 会写成 UPSERT，DELETE 和 UPDATE_BEFORE 会写成 DELETE。

sequence_number_column

sequence_number_column 是可选配置。

当配置了 sequence_number_column 时，该列的值会作为 _CHANGE_SEQUENCE_NUMBER 发送到 BigQuery，用于启用 BigQuery 侧的去重。在 source 重新发送数据时，具有相同 primary key 和相同 sequence number 的行可以由 BigQuery 进行去重。如果没有配置 sequence_number_column，则不会发送 _CHANGE_SEQUENCE_NUMBER，BigQuery 也不会执行基于 sequence number 的去重。

注意
sequence_number_column 应该引用 source 表中单调递增的列，例如以 epoch millis 表示的 updated_at、version 或 seq_id。该列的值必须能够转换为 long 类型。
如果要在 streaming 模式下启用 BigQuery 侧的去重，目标 BigQuery 表必须定义 Primary Key。否则，无论是否配置 sequence number，BigQuery 都会将每次写入视为 append 操作。

emulator_host

emulator_host 只用于本地测试或 CI 测试。配置该参数后，SeaTunnel 会无凭据连接 BigQuery emulator。生产任务不要使用该参数。

任务示例

简单批处理示例

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    row.num = 10
    string.fake.mode = "template"
    string.template = ["key", "value"]
    schema = {
      fields {
        c_map = "map<string, string>"
        c_array = "array<int>"
        c_string = string
        c_boolean = boolean
        c_tinyint = tinyint
        c_smallint = smallint
        c_int = int
        c_bigint = bigint
        c_float = float
        c_double = double
        c_decimal = "decimal(30, 8)"
        c_bytes = bytes
        c_date = date
        c_timestamp = timestamp
        c_time = time
      }
    }
  }
}

sink {
  BigQuery {
    project_id = "test-project"
    dataset_id = "test_dataset"
    table_id = "test_table"
    batch_size = 2
    emulator_host = "localhost:9050"
  }
}

CDC 流式模式（MySQL 到 BigQuery)

目标 BigQuery 表需要提前创建，并且应定义 CDC 源表使用的主键。例如：

CREATE TABLE `my-gcp-project.cdc_dataset.orders` (
  uuid INT64 NOT NULL,
  name STRING,
  score INT64,
  PRIMARY KEY (uuid) NOT ENFORCED
)
OPTIONS (max_staleness = INTERVAL 0 MINUTE);

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 10000
}

source {
  MySQL-CDC {
      parallelism = 1
      server-id = 5652
      username = "st_user_source"
      password = "mysqlpw"
      table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"]
      url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
  }
}

sink {
  BigQuery {
    project_id = "my-gcp-project"
    dataset_id = "cdc_dataset"
    table_id = "orders"
    service_account_key_path = "/path/to/key.json"
    write_mode = "streaming"
    batch_size = 500
  }
}

复杂数据类型示例

source {
  FakeSource {
    row.num = 100
    schema = {
      fields {
        order_id = "bigint"
        customer = {
          name = "string"
          email = "string"
        }
        items = "array<string>"
        metadata = "map<string, string>"
        order_date = "date"
      }
    }
  }
}

sink {
  BigQuery {
    project_id = "my-gcp-project"
    dataset_id = "orders"
    table_id = "customer_orders"
    service_account_key_path = "/path/to/key.json"
    batch_size = 500
  }
}

测试

该连接器使用 BigQuery Storage Write API。当前本地 BigQuery emulator 不能完整支持该连接器使用的写入路径。 emulator_host 只适合用于本地或 CI 中与 emulator 兼容的检查。生产可用性验证应在真实 BigQuery 环境中完成。

更新日志

Change Log

Change	Commit	Version

BigQuery

支持的引擎​

主要特性​

描述​

支持的数据源信息​

配置选项​

认证参数​

表选项​

写入模式​

sequence_number_column​

emulator_host​

任务示例​

简单批处理示例​

CDC 流式模式（MySQL 到 BigQuery)​

复杂数据类型示例​

测试​

更新日志​