企业官网建设流程全解析

SeaTunnel GoogleFirestore Sink 连接器实战指南：将 SeaTunnel 数据行写入 Google Cloud Firestore

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel

本文围绕 SeaTunnel 的connector-google-firestoreSink 连接器展开，系统讲解如何将 SeaTunnel 作业中的每一条数据行（SeaTunnel Row）转换为 Google Cloud Firestore 文档并写入指定 Collection。文章从配置选项、认证方式、数据类型映射到批/流两种作业形态的完整配置示例，并结合仓库源码（FirestoreSinkWriter、FirestoreSinkFactory、DefaultSeaTunnelRowSerializer）剖析其底层实现原理与使用边界。读完本文，你将能够独立完成 GoogleFirestore Sink 的认证配置、字段类型映射校验与批/流作业编写，并理解其"每行一次 Firestore add 调用"写入模型的限制。

一、连接器概述

GoogleFirestore Sink 连接器的作用是把 SeaTunnel 作业产生的数据行写入 Google Cloud 的 Firestore 数据库集合（Collection）。

其核心工作方式是：

• 一行一文档：每一条 SeaTunnel Row 被序列化为一个 Firestore 文档（Document）；
• 自动生成文档 ID：连接器通过调用 Firestore 客户端的add(...)方法写入文档，因此文档 ID 由 Firestore 自动生成，属于追加写入模式，而不是按用户指定的文档 ID 进行更新；
• 凭据灵活：既可以在配置中显式传入 Base64 编码的 Service Account JSON，也可以依赖运行环境的 Google Application Default Credentials（ADC）；
• 不管理索引：连接器不会创建或管理 Firestore 索引，运行涉及索引的查询前需要先在 Google Cloud 控制台创建所需索引。

支持的引擎

功能特性支持矩阵

关于上述特性的详细定义，可参考 Connector V2 特性说明。

依赖获取

依赖可通过install-plugin.sh脚本安装，或从 Maven Central 仓库下载。在插件映射文件 plugin-mapping.properties 中，该连接器的注册信息为seatunnel.sink.GoogleFirestore = connector-google-firestore，对应仓库中的模块为 connector-google-firestore，其 POM 中声明的 Firestore 客户端版本为3.7.10（见 pom.xml）。

二、选项（Options）详解

连接器支持的全部配置项如下：

project_id [string]

必填项。指定拥有 Firestore 数据库的 Google Cloud 项目 ID，值不能为空白字符串。

collection [string]

必填项。指定要写入的 Firestore Collection 名称，值不能为空白字符串。一个 Sink 块只写入一个 Collection——若上游输入来自多个表，需要为每个 Firestore Collection 配置一个独立的 Sink 块。

credentials [string]

可选项。Base64 编码的 Google Cloud Service Account JSON。若配置，值不能为空白字符串。

• 若不配置该选项，连接器使用 Google Application Default Credentials（ADC）：请确保环境变量GOOGLE_APPLICATION_CREDENTIALS指向 Service Account JSON 文件，或运行环境本身已提供默认凭据；
• 生成 Base64 值的命令：

# Linux / 通用 base64 -w 0 service-account.json

# macOS base64 service-account.json | tr -d '\n'

⚠️ 不要将原始的 Service Account JSON 直接填入credentials，必须先做 Base64 编码。

common options

Sink 插件的公共参数，请参阅 Sink Common Options。

配置校验的源码级印证

选项的必填/可选规则在 FirestoreSinkFactory.java 中通过OptionRule定义：

return OptionRule.builder() .required(PROJECT_ID, notBlank(PROJECT_ID)) .required(COLLECTION, notBlank(COLLECTION)) .optional(CREDENTIALS, notBlank(CREDENTIALS)) .build();

即project_id与collection为必填且notBlank（不能为空白），credentials为可选但一旦配置同样要求非空白。三个选项在 FirestoreSinkOptions.java 中定义，均为stringType()且noDefaultValue()。

对应的单元测试 FirestoreFactoryTest.java 覆盖了以下校验场景：

• 缺少project_id或collection时抛出OptionValidationException；
• project_id、collection、credentials传入空串、空格、\t、\n、\r等空白值时均被拒绝；
• 传入未定义的unknown_option会被拒绝（validateUnknownKeys校验）。

三、认证方式与客户端初始化原理

连接器的认证逻辑位于 FirestoreSinkWriter.java 的构造函数中：

GoogleCredentials credentials; if (parameters.getCredentials() != null) { byte[] bytes = Base64.getDecoder().decode(parameters.getCredentials()); credentials = GoogleCredentials.fromStream(new ByteArrayInputStream(bytes)); } else { credentials = GoogleCredentials.getApplicationDefault(); } FirestoreOptions firestoreOptions = FirestoreOptions.getDefaultInstance() .toBuilder() .setProjectId(parameters.getProjectId()) .setCredentials(credentials) .build(); this.firestore = firestoreOptions.getService(); this.collectionReference = firestore.collection(parameters.getCollection()); this.serializer = new DefaultSeaTunnelRowSerializer(seaTunnelRowType);

从源码可以看出完整的初始化链路：

1. 凭据解析：若配置了credentials，先进行 Base64 解码，再通过GoogleCredentials.fromStream(...)从字节流中加载 Service Account；否则回退到GoogleCredentials.getApplicationDefault()读取环境中的 ADC；
2. 构建客户端：基于FirestoreOptions.getDefaultInstance()设置projectId与credentials，随后调用getService()创建 Firestore 客户端；
3. 绑定集合：通过firestore.collection(parameters.getCollection())获取目标 CollectionReference；
4. 准备序列化器：用上游表的SeaTunnelRowType初始化DefaultSeaTunnelRowSerializer，后续每行数据都由它转换为 Firestore 的Map<String, Object>文档字段。

此外，参数对象 FirestoreParameters.java 的buildWithConfig会把作业配置中的project_id、collection、credentials提取为可序列化的参数对象，供 Writer 使用。

四、写入模型：逐行 add 与无缓冲语义

FirestoreSinkWriter的核心写入方法如下：

@Override public void write(SeaTunnelRow seaTunnelRow) throws IOException { collectionReference.add(serializer.serialize(seaTunnelRow)); }

当前实现中write()对每一行调用一次 Firestore 客户端的add(...)方法，既不缓冲也不批量合并行。这意味着：

• 写入请求是逐条同步发起的（每个add返回一个ApiFuture，代码并未显式等待或聚合）；
• 没有内存中的写入缓冲区，因此不存在在 checkpoint 边界进行 flush 的机制；
• checkpoint 完成并不代表此前写入的行已经全部到达 Firestore——这是在使用 checkpoint/容错能力时务必注意的语义差别；
• 该连接器在当前实现下不支持 exactly-once 语义，与第一节特性矩阵中的标注一致。

在close()时，连接器会关闭 Firestore 客户端并释放资源；若关闭失败，会抛出带错误码FIRESTORE-01（"Close Firestore client failed"）的FirestoreConnectorException，错误码定义见 FirestoreConnectorErrorCode.java。

五、字段类型映射（SeaTunnel 类型 → Firestore 类型）

连接器将 SeaTunnel 类型转换为 Firestore 文档字段值，完整映射关系如下表：

序列化的源码实现

上述映射在 DefaultSeaTunnelRowSerializer.java 的convert方法中逐类型实现，几个值得注意的细节：

• TINYINT/SMALLINT/INT统一转为intValue()，BIGINT转为longValue()；
• FLOAT与DOUBLE都转为double写入；
• DECIMAL以BigDecimal原样写入（Firestore 支持 decimal value）；
• BYTES通过Blob.fromBytes(...)转为 Firestore 的 Blob；
• DATE以LocalDate.atStartOfDay(ZoneOffset.UTC)转换为 UTC 当日零点的Date；
• TIMESTAMP以Timestamp.of(...)转为 Firestore Timestamp；
• ARRAY递归转换每个元素为List<Object>，MAP递归转换每个 Value 为对应的 Firestore 值；
• 字段值为null时直接映射为 null。

序列化时上游 SeaTunnel Schema 的字段名会直接成为 Firestore 文档的字段名（见serialize方法中data.put(seaTunnelRowType.getFieldName(index), ...)的逻辑）。

六、使用边界与注意事项（Notes）

以下限制直接决定作业设计与数据模型规划，请务必在开发前评估：

1. 仅提供 Sink：当前连接器只有写入端，没有 GoogleFirestore Source 连接器；
2. 单 Sink 单集合：每个 Sink 块写入一个配置好的 Collection，不会针对多表输入自动切换集合——多集合写入请为每个集合配置一个 Sink 块；
3. 文档 ID 自动生成：由于使用add追加写入，文档 ID 由 Firestore 自动生成。如需确定性文档 ID，请在进入本 Sink 之前使用其他连接器或 Transform 预处理；
4. 不识别 CDC 语义：连接器不会把UPDATE/DELETE行类型解释为 CDC 操作——每一行都会触发一次 Firestoreadd，产生一个新文档；
5. 凭据必须 Base64：不要把原始 Service Account JSON 直接放在credentials中；
6. 字段名继承：上游 SeaTunnel Schema 的字段名会成为 Firestore 文档字段名；
7. checkpoint 语义：连接器同时支持BATCH与STREAMING两种作业模式，但由于write()逐行调用add且无内存缓冲，checkpoint 完成不意味着之前所有行已成功写入 Firestore。

七、任务示例（Task Example）

7.1 批处理写入：全类型字段验证

以下配置使用FakeSource构造一条包含全部支持数据类型的行，并通过 GoogleFirestore Sink 写入：

env { parallelism = 1 job.mode = "BATCH" } source { FakeSource { 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_null = "null" c_bytes = bytes c_date = date c_timestamp = timestamp } } rows = [ { kind = INSERT fields = [{"a": "b"}, [10], "c_string", true, 117, 15987, 56387395, 7084913402530365000, 1.23, 1.23, "2924137191386439303744.39292216", null, "bWlJWmo=", "2023-04-22", "2023-04-22T23:20:58"] } ] } } sink { GoogleFirestore { project_id = "dummy-project" collection = "dummy-collection" credentials = "base64-service-account-json" } }

该配置与仓库 E2E 测试使用的 fake_to_google_firestore.conf 一致。对应的端到端测试 GoogleFirestoreIT.java 会执行作业后读取 Firestore，断言写入文档字段数为 15，并逐项校验15987L（smallint）、56387395L（int）、"2924137191386439303744.39292216"（decimal）、Blob.fromBytes(...)（bytes）以及 Timestamp 等转换结果。该测试默认@Disabled，因为需要真实的 Google Firestore 数据库环境才能运行。

7.2 流式写入：配合 checkpoint 间隔

以下配置演示STREAMING模式下、指定 checkpoint 间隔的写入：

env { parallelism = 1 job.mode = "STREAMING" checkpoint.interval = 30000 } source { FakeSource { row.num = 100 schema = { fields { c_string = string c_int = int c_timestamp = timestamp } } plugin_output = "firestore_stream" } } sink { GoogleFirestore { plugin_input = "firestore_stream" project_id = "my-gcp-project" collection = "events" credentials = "base64-service-account-json" } }

示例要点：

• job.mode = "STREAMING"且checkpoint.interval = 30000（单位毫秒），开启 30 秒周期的 checkpoint；
• 通过plugin_output/plugin_input显式串联 FakeSource 与 GoogleFirestore Sink；
• 结合第四节所述的无缓冲逐行add模型，请注意 checkpoint 不代表数据已达 Firestore。

八、使用流程小结

1. 准备凭据：在 Google Cloud 创建 Service Account 并下载 JSON，执行base64 -w 0 service-account.json（macOS 用base64 service-account.json | tr -d '\n'）得到配置值；或确保运行环境可解析 Application Default Credentials（如设置GOOGLE_APPLICATION_CREDENTIALS）；
2. 确认目标：确定project_id与目标 Collection，并按需在 Google Cloud 预先创建涉及查询的索引；
3. 编写作业配置：按第七节模板配置env/source/sink，其中 Sink 块至少包含project_id与collection，可选credentials；
4. 运行与验证：使用 SeaTunnel 引擎（Zeta、Spark 或 Flink）提交作业，可在 Firestore 控制台或通过客户端查询确认文档已写入、类型映射符合第五节表格。

九、版本演进参考

依据 connector-google-firestore 变更日志：

• 2.3.2：新增 GoogleFirestore Sink 连接器（Feature）；
• 2.3.4：移除对SeaTunnelSink::getConsumedType的使用并标记废弃；
• 2.3.9：允许将指标信息关联到逻辑计划节点；
• 2.3.10：改进 Firestore 选项。

如需进一步了解 Sink 公共选项，请阅读 Sink Common Options；若想基于本文示例扩展更多连接器用法，可参考仓库中的 connector-google-firestore 模块 及其 E2E 测试。

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel