Airbyte 目的地连接器开发指南（四）：Write Operations 写入操作与同步核心功能实现

数据工程数据集成ETL后端大数据【免费下载链接】airbyteOpen-source data movement for ELT pipelines and AI agents — from APIs, databases files to warehouses, lakes, and AI applications. Both self-hosted and Cloud.项目地址https://gitcode.com/gh_mirrors/ai/airbyte点击查看免费下载导读本文是 Airbyte 开源仓库中「逐步构建 Java/Kotlin 目的地连接器」系列教程的第四篇对应仓库文档 4-write-operations.md。在完成 3-write-infrastructure.mdDI 基础设施与测试上下文之后本指南将带你实现连接器真正「写入数据」的业务逻辑InsertBuffer 批量缓冲、Aggregate 记录汇聚、Writer 编排、Append 追加模式、Overwrite 全量覆盖模式、Generation ID 同步代次追踪以及支撑上述能力的ConnectorWiringSuite组件测试。读完本文你将掌握--write操作在基础同步场景下的完整实现链路并能用 Gradle 命令逐阶段验证每个里程碑。一、Write Phase 1Writer 与 Append 模式核心业务逻辑1.1 阶段目标与依赖上下文本阶段的目标是实现真实的数据写入核心组件为 Writer、Aggregate 与 InsertBuffer。在继续之前请确保前一篇文档中的基础设施WriteOperationV2、DatabaseInitialStatusGatherer、ColumnNameMapper、名称生成器已全部就绪并注册为 Micronaut Bean。一个贯穿整个系列的关键洞察是基础设施 DIPhase 7与业务逻辑 DIPhase 8是分离的。Phase 7 验证的是「连接器能否启动」Phase 8 验证的则是「连接器能否写入数据」。把两者分开可以在开发过程中增量地捕获错误而不是等到最后一次性暴露。本阶段完成后你将拥有以下能力InsertBuffer批量累积记录并刷写到数据库Aggregate处理经过转换transform后的记录AggregateFactory按流stream创建 Aggregate 实例Writer编排启动流程并创建 StreamLoaderCheckpoint能够端到端写入一条记录。1.2 创建 InsertBuffer按流缓冲、按阈值刷写InsertBuffer 的职责是累积记录、在达到阈值时批量刷写。文件位于write/load/{DB}InsertBuffer.ktpackage io.airbyte.integrations.destination.{db}.write.load import io.airbyte.cdk.load.data.AirbyteValue import io.airbyte.cdk.load.schema.model.TableName import io.airbyte.integrations.destination.{db}.client.{DB}AirbyteClient import io.github.oshai.kotlinlogging.KotlinLogging private val log KotlinLogging.logger {} /** * Accumulates records and flushes to database in batches. * * NOT a Singleton - created per-stream by AggregateFactory */ class {DB}InsertBuffer( private val tableName: TableName, private val client: {DB}AirbyteClient, private val flushLimit: Int 1000, ) { private val buffer mutableListOfMapString, AirbyteValue() private var recordCount 0 fun accumulate(recordFields: MapString, AirbyteValue) { buffer.add(recordFields) recordCount if (recordCount flushLimit) { kotlinx.coroutines.runBlocking { flush() } } } suspend fun flush() { if (buffer.isEmpty()) return try { log.info { Flushing $recordCount records to ${tableName}... } // Simple multi-row INSERT for now // (Optimize in Phase 15: CSV staging, COPY, bulk APIs) buffer.forEach { record - insertRecord(tableName, record) } log.info { Finished flushing $recordCount records } } finally { buffer.clear() recordCount 0 } } private suspend fun insertRecord( tableName: TableName, record: MapString, AirbyteValue ) { val columns record.keys.joinToString(, ) { \$it\ } val placeholders record.keys.joinToString(, ) { ? } val sql INSERT INTO ${tableName.namespace}.${tableName.name} ($columns) VALUES ($placeholders) client.executeInsert(sql, record.values.toList()) } }几个要点绝对不是Singleton——每条流stream都必须拥有自己独立的 buffer因为 buffer 持有流级状态目标表名、累积的记录。AggregateFactory会为每条流创建一个 buffer。当前是简单实现逐行 INSERT。批量加载CSV staging、COPY、批量 API留待后续优化阶段Phase 15处理。flushLimit默认 1000 条达到阈值后自动触发flush()flush()在finally块中清空 buffer保证无论成功失败都不会残留状态。TableName与AirbyteValue均来自 CDK 的io.airbyte.cdk.load包schema.model与data。仓库中的真实实现可以印证这一设计例如 ClickHouseWriter.kt 同样遵循「Writer 编排 按流管理写入」的结构。1.3 在 Client 中增加 executeInsert() 与参数绑定JDBC 场景下需要在client/{DB}AirbyteClient.kt中添加executeInsert()与参数绑定方法// Add this method to {DB}AirbyteClient fun executeInsert(sql: String, values: ListAirbyteValue) { dataSource.connection.use { connection - connection.prepareStatement(sql).use { statement - values.forEachIndexed { index, value - setParameter(statement, index 1, value) } statement.executeUpdate() } } } private fun setParameter(statement: PreparedStatement, index: Int, value: AirbyteValue) { when (value) { is StringValue - statement.setString(index, value.value) is IntegerValue - statement.setLong(index, value.value) is NumberValue - statement.setBigDecimal(index, value.value) is BooleanValue - statement.setBoolean(index, value.value) is TimestampValue - statement.setTimestamp(index, Timestamp.from(value.value)) is DateValue - statement.setDate(index, Date.valueOf(value.value)) is TimeValue - statement.setTime(index, Time.valueOf(value.value.toLocalTime())) is ObjectValue - statement.setString(index, value.toJson()) // JSON as string is ArrayValue - statement.setString(index, value.toJson()) // JSON as string is NullValue - statement.setNull(index, Types.VARCHAR) else - statement.setString(index, value.toString()) } }参数绑定遵循AirbyteValue各子类型的语义整数用setLong、数字用setBigDecimal保证精度、时间戳用setTimestamp对象与数组以 JSON 字符串形式落库。注意对于非 JDBC 数据库应改用原生客户端 API例如 MongoDB 的insertOne、ClickHouse 原生客户端。1.4 创建 Aggregate接入 CDK 数据流管线Aggregate 是 CDK 数据流管线的最终落点。文件位于dataflow/{DB}Aggregate.ktpackage io.airbyte.integrations.destination.{db}.dataflow import io.airbyte.cdk.load.dataflow.aggregate.Aggregate import io.airbyte.cdk.load.dataflow.transform.RecordDTO import io.airbyte.integrations.destination.{db}.write.load.{DB}InsertBuffer /** * Processes transformed records for a single stream. * * Dataflow pipeline: Raw record → Transform → RecordDTO → Aggregate.accept() → InsertBuffer * * NOT a Singleton - created per-stream by AggregateFactory */ class {DB}Aggregate( private val buffer: {DB}InsertBuffer, ) : Aggregate { override fun accept(record: RecordDTO) { buffer.accumulate(record.fields) } override suspend fun flush() { buffer.flush() } }它实现的是 CDK 提供的io.airbyte.cdk.load.dataflow.aggregate.Aggregate接口。从源码 Aggregate.kt 可以看到该接口仅有两个方法fun accept(record: RecordDTO)把记录加入当前批次字段与值的转换如 ValueCoercer已在更上游完成suspend fun flush()由 CDK 在批次达到AggregatePublishingConfig阈值时自动调用执行真正的批量加载。CDK 会在内部根据 AggregatePublishingConfig 的配置maxRecordsPerAgg、maxEstBytesPerAgg、maxEstBytesAllAggregates、maxBufferedAggregates自动决定何时触发 flush——这就是上一阶段在 BeanFactory 中注册该配置的意义所在。完整数据流管线如下这也是理解整篇教程的脉络图Platform → JSONL records ↓ AirbyteMessageDeserializer (CDK) ↓ RecordTransformer (CDK, uses ColumnNameMapper from Phase 7) ↓ RecordDTO (transformed record with mapped column names) ↓ Aggregate.accept() ← YOUR CODE STARTS HERE ↓ InsertBuffer.accumulate() ↓ Database1.5 创建 AggregateFactory工厂模式解决动态流问题由于流列表是动态的、无法用构造函数注入需要工厂模式按StoreKey查表创建 Aggregate。文件位于dataflow/{DB}AggregateFactory.ktpackage io.airbyte.integrations.destination.{db}.dataflow import io.airbyte.cdk.load.dataflow.aggregate.Aggregate import io.airbyte.cdk.load.dataflow.aggregate.AggregateFactory import io.airbyte.cdk.load.state.StoreKey import io.airbyte.cdk.load.table.directload.DirectLoadTableExecutionConfig import io.airbyte.cdk.load.write.StreamStateStore import io.airbyte.integrations.destination.{db}.client.{DB}AirbyteClient import io.airbyte.integrations.destination.{db}.write.load.{DB}InsertBuffer import io.micronaut.context.annotation.Factory import jakarta.inject.Singleton Factory class {DB}AggregateFactory( private val client: {DB}AirbyteClient, private val streamStateStore: StreamStateStoreDirectLoadTableExecutionConfig, ) : AggregateFactory { Singleton override fun create(key: StoreKey): Aggregate { // StreamStateStore contains execution config for each stream // Config includes table name, column mapping, etc. val tableName streamStateStore.get(key)!!.tableName val buffer {DB}InsertBuffer( tableName tableName, client client, ) return {DB}Aggregate(buffer) } }要点Factory类提供创建 Aggregate 的工厂方法create()在每次同步开始时按流调用一次。StreamStateStore保存每条流的执行配置表名等。仓库中的 StreamStateStore.kt 就是这一契约的实现。为什么需要工厂模式Aggregate 需要流级状态表名无法使用构造注入流列表是动态的因此工厂接收StoreKey、查询流配置、再创建 Aggregate。CDK 的AggregateFactory接口签名fun create(key: StoreKey): Aggregate在 Aggregate.kt 中同样可以找到。1.6 创建 Writer编排启动与 StreamLoader 选择Writer 是整个写入流程的编排者。文件位于write/{DB}Writer.ktpackage io.airbyte.integrations.destination.{db}.write import io.airbyte.cdk.SystemErrorException import io.airbyte.cdk.load.command.DestinationCatalog import io.airbyte.cdk.load.command.DestinationStream import io.airbyte.cdk.load.table.ColumnNameMapping import io.airbyte.cdk.load.table.DatabaseInitialStatusGatherer import io.airbyte.cdk.load.table.directload.DirectLoadInitialStatus import io.airbyte.cdk.load.table.directload.DirectLoadTableAppendStreamLoader import io.airbyte.cdk.load.table.directload.DirectLoadTableAppendTruncateStreamLoader import io.airbyte.cdk.load.table.directload.DirectLoadTableExecutionConfig import io.airbyte.cdk.load.write.DestinationWriter import io.airbyte.cdk.load.write.StreamLoader import io.airbyte.cdk.load.write.StreamStateStore import io.airbyte.integrations.destination.{db}.client.{DB}AirbyteClient import jakarta.inject.Singleton Singleton class {DB}Writer( private val catalog: DestinationCatalog, private val stateGatherer: DatabaseInitialStatusGathererDirectLoadInitialStatus, private val streamStateStore: StreamStateStoreDirectLoadTableExecutionConfig, private val client: {DB}AirbyteClient, ) : DestinationWriter { private lateinit var initialStatuses: MapDestinationStream, DirectLoadInitialStatus override suspend fun setup() { // Create all namespaces catalog.streams .map { it.tableSchema.tableNames.finalTableName!!.namespace } .toSet() .forEach { client.createNamespace(it) } // Gather initial state (which tables exist, generation IDs, etc.) initialStatuses stateGatherer.gatherInitialStatus() } override fun createStreamLoader(stream: DestinationStream): StreamLoader { val initialStatus initialStatuses[stream]!! // Access schema directly from stream (modern CDK pattern) val realTableName stream.tableSchema.tableNames.finalTableName!! val tempTableName stream.tableSchema.tableNames.tempTableName!! val columnNameMapping ColumnNameMapping( stream.tableSchema.columnSchema.inputToFinalColumnNames ) // Choose StreamLoader based on sync mode return when (stream.minimumGenerationId) { 0L - // Append mode: just insert records DirectLoadTableAppendStreamLoader( stream, initialStatus, realTableName realTableName, tempTableName tempTableName, columnNameMapping, client, // TableOperationsClient client, // TableSchemaEvolutionClient streamStateStore, ) stream.generationId - // Overwrite/truncate mode: replace table contents DirectLoadTableAppendTruncateStreamLoader( stream, initialStatus, realTableName realTableName, tempTableName tempTableName, columnNameMapping, client, client, streamStateStore, ) else - throw SystemErrorException( Cannot execute a hybrid refresh - current generation ${stream.generationId}; minimum generation ${stream.minimumGenerationId} ) } } }Writer 的职责拆分setup()为 catalog 中所有流的命名空间namespace建库/建 schema并通过DatabaseInitialStatusGatherer收集初始状态哪些表已存在、存在哪个 generation ID 等。createStreamLoader()按同步模式为每条流创建对应的 StreamLoader。现代 CDK 模式stream.tableSchemaschema 信息已经由 CDK 内嵌在stream.tableSchema中直接通过stream.tableSchema.tableNames.finalTableName!!和stream.tableSchema.columnSchema.inputToFinalColumnNames访问即可不需要防御性空检查——CDK 保证 schema 存在。StreamLoader 选择逻辑minimumGenerationId语义minimumGenerationId模式StreamLoader0LAppend 追加DirectLoadTableAppendStreamLoader stream.generationIdOverwrite 覆盖DirectLoadTableAppendTruncateStreamLoader其他组合不支持抛出SystemErrorExceptionhybrid refresh 不支持仓库中这四种 StreamLoader 的真实实现位于 DirectLoadTableStreamLoader.kt第 35、89、154、293 行分别对应 Append、Dedup、AppendTruncate、DedupTruncate 四种加载器。以DirectLoadTableAppendStreamLoader为例其start()方法会若目标表不存在则createTable(replace false)非截断模式禁止replacetrue防止误删已有数据若表已存在则调用ensureSchemaMatches()做 schema 演化若存在残留的临时表来自上次失败同步则先copyTable数据再dropTable最后把DirectLoadTableExecutionConfig写入streamStateStore。CDK 提供的四种实现分别为DirectLoadTableAppendStreamLoader、DirectLoadTableAppendTruncateStreamLoader、DirectLoadTableDedupStreamLoader、DirectLoadTableDedupTruncateStreamLoader。StreamLoader 的生命周期契约是start()创建/准备表 →accept()加入缓冲 →complete()刷写并收尾。1.7 创建 ConnectorWiringSuite 测试测试文件位于src/test-integration/kotlin/.../component/{DB}WiringTest.ktpackage io.airbyte.integrations.destination.{db}.component import io.airbyte.cdk.load.component.ConnectorWiringSuite import io.airbyte.cdk.load.component.TableOperationsClient import io.airbyte.cdk.load.dataflow.aggregate.AggregateFactory import io.airbyte.cdk.load.write.DestinationWriter import io.micronaut.test.extensions.junit5.annotation.MicronautTest import org.junit.jupiter.api.Test MicronautTest(environments [component]) class {DB}WiringTest( override val writer: DestinationWriter, override val client: TableOperationsClient, override val aggregateFactory: AggregateFactory, ) : ConnectorWiringSuite { // Optional: Override test namespace if different from test // override val testNamespace my_database Test override fun all beans are injectable() { super.all beans are injectable() } Test override fun writer setup completes() { super.writer setup completes() } Test override fun can create append stream loader() { super.can create append stream loader() } Test override fun can write one record() { super.can write one record() } }ConnectorWiringSuite接口定义于 ConnectorWiringSuite.kt它要求测试类通过 Micronaut 注入三个核心 BeanDestinationWriter、TableOperationsClient、AggregateFactory。四个测试用例的职责如下测试 1all beans are injectable验证所有 DI Bean 可注入捕获缺失的Singleton注解、循环依赖等问题。测试 2writer setup completes调用Writer.setup()验证命名空间创建可用捕获数据库连接错误。测试 3can create append stream loader调用Writer.createStreamLoader()验证 StreamLoader 实例化捕获缺失的 StreamLoader 依赖。测试 4can write one record最重要端到端验证完整写入路径创建测试流调用StreamLoader.start()→ 建表调用Aggregate.accept()→ 缓冲记录调用Aggregate.flush()→ 写入数据库验证记录出现在数据库中。测试上下文使用MockDestinationCatalog快速迭代无需解析真实 catalog JSON可动态创建测试流聚焦写入逻辑而非 catalog 解析数据库使用 Testcontainers属于组件测试component test而非集成测试。1.8 验证 Phase 1$ ./gradlew :destination-{db}:testComponentAllBeansAreInjectable \ :destination-{db}:testComponentWriterSetupCompletes \ :destination-{db}:testComponentCanCreateAppendStreamLoader \ :destination-{db}:testComponentCanWriteOneRecord # 4 tests should pass $ ./gradlew :destination-{db}:componentTest # 9 tests should pass $ ./gradlew :destination-{db}:integrationTest # 3 tests should pass如果can write one record失败按以下清单排查错误类别检查点DI 错误检查 Phase 7 基础设施WriteOperationV2、DatabaseInitialStatusGatherer、ColumnNameMapper检查 Phase 6 名称生成器都有Singleton建表错误检查TableOperationsClient.createTable()Phase 4检查SqlGenerator.createTable()的 SQL 语法插入错误检查InsertBuffer.insertRecord()实现检查client.executeInsert()与setParameter()逻辑检查列名映射记录未写入检查buffer.flush()是否被调用检查 INSERT SQL 是否正确直接查询数据库调试✅Checkpoint首个可工作的同步 之前所有阶段仍然通过。二、Write Phase 2Generation ID 支持2.1 什么是 Generation ID**Generation ID代次标识**是每次同步运行的唯一标识用于在全量刷新refresh时区分「旧数据」与「新数据」存储于_airbyte_generation_id列中。其使用规则为全量刷新Full refreshminimumGenerationId generationId替换所有数据增量同步IncrementalminimumGenerationId 0保留所有数据2.2 启用 Generation ID 测试在src/test-integration/kotlin/.../component/{DB}TableOperationsTest.kt中启用测试Test override fun get generation id() { super.get generation id() }该测试验证三件事TableOperationsClient.getGenerationId()返回正确的值对没有 generation ID 的表返回0L能从_airbyte_generation_id列读到实际的 generation ID。2.3 验证 Phase 2$ ./gradlew :destination-{db}:testComponentGetGenerationId # 1 test should pass $ ./gradlew :destination-{db}:componentTest # 10 tests should pass✅CheckpointGeneration ID 追踪生效 之前所有阶段仍然通过。三、Write Phase 3Overwrite 全量覆盖模式3.1 覆盖模式的原子换表机制Overwrite 的工作方式将新数据写入临时表temp table原子性地将临时表与正式表交换SWAP删除旧表。同步模式对比AppendPhase 8INSERT 进已存在的表OverwritePhase 10SWAP 临时表与正式表。3.2 在 SQL Generator 中实现 overwriteTable()在client/{DB}SqlGenerator.kt中实现文档给出了四种数据库适配方案fun overwriteTable(source: TableName, target: TableName): ListString { // Option 1: SWAP (Snowflake) return listOf( ALTER TABLE ${fullyQualifiedName(target)} SWAP WITH ${fullyQualifiedName(source)}.andLog(), DROP TABLE IF EXISTS ${fullyQualifiedName(source)}.andLog(), ) // Option 2: EXCHANGE (ClickHouse) return listOf( EXCHANGE TABLES ${fullyQualifiedName(target)} AND ${fullyQualifiedName(source)}.andLog(), DROP TABLE IF EXISTS ${fullyQualifiedName(source)}.andLog(), ) // Option 3: DROP RENAME (fallback for most databases) return listOf( DROP TABLE IF EXISTS ${fullyQualifiedName(target)}.andLog(), ALTER TABLE ${fullyQualifiedName(source)} RENAME TO ${target.name.quote()}.andLog(), ) // Option 4: BEGIN TRANSACTION DROP RENAME COMMIT (for ACID guarantees) return listOf( BEGIN TRANSACTION.andLog(), DROP TABLE IF EXISTS ${fullyQualifiedName(target)}.andLog(), ALTER TABLE ${fullyQualifiedName(source)} RENAME TO ${target.name.quote()}.andLog(), COMMIT.andLog(), ) }数据库差异速查数据库方案说明SnowflakeSWAP原子且瞬时元数据操作ClickHouseEXCHANGE原子Postgres / MySQLDROP RENAME需要事务包裹才能保证原子性BigQueryCREATE OR REPLACE TABLE不同模式.andLog()是 CDK SQL Generator 的惯用扩展生成 SQL 的同时打印日志便于排查。3.3 在 Client 中实现 overwriteTable()override suspend fun overwriteTable( sourceTableName: TableName, targetTableName: TableName ) { val statements sqlGenerator.overwriteTable(sourceTableName, targetTableName) statements.forEach { execute(it) } }3.4 更新 Writer 支持 Truncate 模式与 1.6 节中的版本相比主要变化是增加了对不在 catalog 中流的防御性处理保证测试兼容并显式区分两种模式的 StreamLoaderoverride fun createStreamLoader(stream: DestinationStream): StreamLoader { // Defensive: Handle streams not in catalog (for test compatibility) val initialStatus if (::initialStatuses.isInitialized) { initialStatuses[stream] ?: DirectLoadInitialStatus(null, null) } else { DirectLoadInitialStatus(null, null) } val tableNameInfo names[stream] val (realTableName, tempTableName, columnNameMapping) if (tableNameInfo ! null) { Triple( tableNameInfo.tableNames.finalTableName!!, tempTableNameGenerator.generate(tableNameInfo.tableNames.finalTableName!!), tableNameInfo.columnNameMapping ) } else { val tableName TableName( namespace stream.mappedDescriptor.namespace ?: test, name stream.mappedDescriptor.name ) Triple(tableName, tempTableNameGenerator.generate(tableName), ColumnNameMapping(emptyMap())) } // Choose StreamLoader based on sync mode return when (stream.minimumGenerationId) { 0L - DirectLoadTableAppendStreamLoader( stream, initialStatus, realTableName, tempTableName, columnNameMapping, client, client, streamStateStore ) stream.generationId - DirectLoadTableAppendTruncateStreamLoader( stream, initialStatus, realTableName, tempTableName, columnNameMapping, client, client, streamStateStore ) else - throw SystemErrorException(Hybrid refresh not supported) } }两种 StreamLoader 的行为差异DirectLoadTableAppendStreamLoader直接写入正式表保留旧数据DirectLoadTableAppendTruncateStreamLoader先写临时表、再交换替换旧数据。对照仓库源码 DirectLoadTableStreamLoader.kt 第 154 行开始的DirectLoadTableAppendTruncateStreamLoader可以看到截断路径有意保留replacetrue因为该模式预期表被整体替换而第 35 行的 Append 加载器使用replacefalse防止意外丢数据。3.5 启用测试并验证在{DB}TableOperationsTest.kt中启用Test override fun overwrite tables() { super.overwrite tables() }验证$ ./gradlew :destination-{db}:testComponentOverwriteTables # 1 test should pass $ ./gradlew :destination-{db}:componentTest # 11 tests should pass $ ./gradlew :destination-{db}:integrationTest # 3 tests should pass✅Checkpoint全量刷新模式可用 之前所有阶段仍然通过。四、Write Phase 4Copy 复制操作4.1 Copy 的适用场景表复制操作被多种模式内部使用Dedupe 去重模式把去重后的数据从临时表复制到正式表部分 Overwrite 实现用复制代替交换Schema 演化复制到新 schema。4.2 在 SQL Generator 中实现 copyTable()fun copyTable( columnMapping: ColumnNameMapping, source: TableName, target: TableName ): String { val columnList columnMapping.values.joinToString(, ) { \$it\ } return INSERT INTO ${fullyQualifiedName(target)} ($columnList) SELECT $columnList FROM ${fullyQualifiedName(source)} .trimIndent().andLog() }该实现的特点是把源表所有行复制到目标表只复制映射过的列而非全部列通过 SELECT → INSERT 保留数据类型。备选方案显式包含 Airbyte 元数据列。如果目标表需要保留 Airbyte 元数据使用以下版本fun copyTable( columnMapping: ColumnNameMapping, source: TableName, target: TableName ): String { // Include Airbyte metadata user columns val allColumns listOf( _airbyte_raw_id, _airbyte_extracted_at, _airbyte_meta, _airbyte_generation_id ) columnMapping.values val columnList allColumns.joinToString(, ) { \$it\ } return INSERT INTO ${fullyQualifiedName(target)} ($columnList) SELECT $columnList FROM ${fullyQualifiedName(source)} .trimIndent().andLog() }4.3 在 Client 中实现 copyTable()override suspend fun copyTable( columnNameMapping: ColumnNameMapping, sourceTableName: TableName, targetTableName: TableName ) { execute(sqlGenerator.copyTable(columnNameMapping, sourceTableName, targetTableName)) }值得说明的是copyTable不只是本阶段的练习——如 1.6 节所析CDK 的DirectLoadTableAppendStreamLoader.start()在检测到上次失败同步残留的临时表时正是调用tableOperationsClient.copyTable(...)把临时表数据救回正式表见 DirectLoadTableStreamLoader.kt因此该能力同时是同步恢复机制的一部分。4.4 启用测试并验证Test override fun copy tables() { super.copy tables() }$ ./gradlew :destination-{db}:testComponentCopyTables # 1 test should pass $ ./gradlew :destination-{db}:componentTest # 12 tests should pass $ ./gradlew :destination-{db}:integrationTest # 3 tests should pass✅CheckpointCopy 操作可用 之前所有阶段仍然通过。五、下一步完成本指南后你的连接器已经支持基础同步场景Append 追加、Overwrite 全量覆盖、Generation ID 追踪与表复制并通过了ConnectorWiringSuite的端到端写入验证。接下来可以继续 5-advanced-features.md 实现生产级特性schema 演化、Dedupe 去重模式、CDC 删除支持、批量写入优化或直接进入 6-testing.md 运行完整测试套件遇到问题时参考 7-troubleshooting.md 中的测试上下文与 DI 错误排查指引并通过 8-validation.md 做最终验收。赞分享数据工程数据集成ETL后端大数据【免费下载链接】airbyteOpen-source data movement for ELT pipelines and AI agents — from APIs, databases files to warehouses, lakes, and AI applications. Both self-hosted and Cloud.项目地址https://gitcode.com/gh_mirrors/ai/airbyte点击查看免费下载相关推荐Airbyte source-mongodb-v2 连接器本地开发与 Bug 复现完全指南Airbyte source mongodb v2 连接器本地开发与 Bug 复现完全指南 source mongodb v2 是 Airbyte 生态中基于数据工程数据集成ETL后端大数据深入解析 Airbyte Fauna Source 连接器从本地开发、同步模式到数据序列化深入解析 Airbyte Fauna Source 连接器从本地开发、同步模式到数据序列化 Airbyte 生态中的 source fauna 是面向 Fau数据工程数据集成ETL后端大数据Airbyte连接器开发实战Airbyte连接器开发实战 本文详细介绍了Airbyte连接器开发的三种主要方式无代码Connector Builder、低代码CDK开发流程以及API和数数据工程数据集成ETL后端大数据创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考