跳到主要内容
版本：Next

Opengauss CDC

Opengauss CDC源连接器

支持这些引擎​

SeaTunnel Zeta
Flink

主要功能​

描述​

Opengauss CDC 连接器允许从 Opengauss 数据库读取快照数据和增量数据。本文档介绍如何配置 Opengauss CDC 连接器。

使用步骤​

这里是启用Opengauss CDC的步骤:

  1. 确保wal_level被设置为logical, 你可以直接使用SQL命令来修改这个配置:
ALTER SYSTEM SET wal_level TO 'logical';
SELECT pg_reload_conf();
  1. 改变指定表的REPLICA策略为FULL
ALTER TABLE your_table_name REPLICA IDENTITY FULL;

如果你有很多表，你可以使用下面SQL的结果集来改变所有表的REPLICA策略

select 'ALTER TABLE ' || schemaname || '.' || tablename || ' REPLICA IDENTITY FULL;' from pg_tables where schemaname = 'YourTableSchema'

数据类型映射​

Opengauss Data typeSeaTunnel Data type
BOOL
BOOLEAN
BYTEA
BYTES
INT2
SMALLSERIAL
INT4
SERIAL
INT
INT8
BIGSERIAL
BIGINT
FLOAT4
FLOAT
FLOAT8
DOUBLE
NUMERIC(Get the designated column's specified column size>0)DECIMAL(Get the designated column's specified column size,Gets the number of digits in the specified column to the right of the decimal point)
NUMERIC(Get the designated column's specified column size<0)DECIMAL(38, 18)
BPCHAR
CHARACTER
VARCHAR
TEXT
GEOMETRY
GEOGRAPHY
JSON
JSONB		STRING
TIMESTAMP
TIMESTAMP
TIME
TIME
DATE
DATE
OTHER DATA TYPESNOT SUPPORTED YET

源端可选项​

NameTypeRequiredDefaultDescription
url字符串是-JDBC连接的URL. 参考: jdbc:postgresql://localhost:5432/postgres_cdc?loggerLevel=OFF.
username字符串是-连接数据库的用户名
password字符串是-连接数据库的密码
database-names列表否-需要监控的数据库名称。
table-names列表二选一-需要监控的表。请使用完整的 database.schema.table 格式，例如：opengauss_cdc.inventory.orders。
table-pattern字符串二选一-需要监控的表名正则表达式。正则需要匹配完整表名，例如：opengauss_cdc\\.inventory\\..*。table-names 和 table-pattern 互斥。
table-names-config列表否-表级配置列表。例如：[{"table": "db1.schema1.table1","primaryKeys": ["key1"],"snapshotSplitColumn": "key2"}]。无物理主键表可通过 primaryKeys 指定唯一键。snapshotSplitColumn 必须是唯一键，否则 SeaTunnel 会忽略该配置并自动选择拆分列。
startup.mode枚举否INITIALOpengauss CDC消费者的可选启动模式, 有效的枚举是initial, earliest, latest.
initial: 启动时同步历史数据，然后同步增量数据
earliest: 从可能的最早偏移量启动
latest: 从最近的偏移量启动
snapshot.split.size整型否8096表快照的分割大小（行数），在读取表的快照时，捕获的表被分割成多个split
snapshot.fetch.size整型否1024读取表快照时，每次轮询的最大读取大小
slot.name字符串否seatunnelOpengauss 逻辑解码槽名称。同一个 Opengauss 实例上如果有多个 CDC 任务，请为每个任务配置不同的 slot.name。
decoding.plugin.name字符串否pgoutput安装在服务器上的Postgres逻辑解码插件的名称，支持的值是decoderbufs、wal2json、wal2json_rds、wal2json_streaming、wal2json_rds_streaming和pgoutput
server-time-zone字符串否UTC数据库服务器中的会话时区。如果没有设置，则使用ZoneId.systemDefault()来确定服务器的时区
connect.timeout.ms时间间隔否30000在尝试连接数据库服务器之后，连接器在超时之前应该等待的最大时间
connect.max-retries整型否3连接器在建立数据库服务器连接时应该重试的最大次数
connection.pool.size整型否20jdbc连接池的大小
chunk-key.even-distribution.factor.upper-bound双浮点型否100chunk的key分布因子的上界。该因子用于确定表数据是否均匀分布。如果分布因子被计算为小于或等于这个上界(即(MAX(id) - MIN(id) + 1) /行数)，表的所有chunk将被优化以达到均匀分布。否则，如果分布因子更大，则认为表分布不均匀，如果估计的分片数量超过sample-sharding.threshold指定的值，则将使用基于采样的分片策略。默认值为100.0。
chunk-key.even-distribution.factor.lower-bound双浮点型否0.05chunk的key分布因子的下界。该因子用于确定表数据是否均匀分布。如果分布因子的计算结果大于或等于这个下界(即(MAX(id) - MIN(id) + 1) /行数)，那么表的所有块将被优化以达到均匀分布。否则，如果分布因子较小，则认为表分布不均匀，如果估计的分片数量超过sample-sharding.threshold指定的值，则使用基于采样的分片策略。缺省值为0.05。
sample-sharding.threshold整型否1000此配置指定了用于触发采样分片策略的估计分片数的阈值。当分布因子超出了由chunk-key.even-distribution.factor.upper-bound 和chunk-key.even-distribution.factor.lower-bound，并且估计的分片计数(以近似的行数/块大小计算)超过此阈值，则将使用样本分片策略。这有助于更有效地处理大型数据集。默认值为1000个分片。
inverse-sampling.rate整型否1000采样分片策略中使用的采样率的倒数。例如，如果该值设置为1000，则意味着在采样过程中应用了1/1000的采样率。该选项提供了控制采样粒度的灵活性，从而影响最终的分片数量。当处理非常大的数据集时，它特别有用，其中首选较低的采样率。缺省值为1000。
split.allow-sampling布尔否true是否允许使用基于采样的分片策略。当设置为 false 时，无论预估分片数是否超过阈值，系统都会回退到非均匀分片方式（迭代查询方式）。默认值为 true。
exactly_once布尔否false在初始快照阶段启用精确一次语义。仅当 startup.mode 为 initial 时可用。
format枚举否DEFAULTOpengauss CDC可选的输出格式, 有效的枚举是DEFAULT, COMPATIBLE_DEBEZIUM_JSON.
debezium配置否-将 Debezium的属性 传递到Debezium嵌入式引擎，该引擎用于捕获来自Opengauss服务的数据更改
common-options否-源码插件通用参数, 请参考Source Common Options获取详情

任务示例​

简单​

支持多表读


env {
  # You can set engine configuration here
  execution.parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 5000
  read_limit.bytes_per_second=7000000
  read_limit.rows_per_second=400
}

source {
  Opengauss-CDC {
    plugin_output = "customers_opengauss_cdc"
    username = "gaussdb"
    password = "openGauss@123"
    database-names = ["opengauss_cdc"]
    table-names = ["opengauss_cdc.inventory.opengauss_cdc_table_1","opengauss_cdc.inventory.opengauss_cdc_table_2"]
    url = "jdbc:postgresql://opengauss_cdc_e2e:5432/opengauss_cdc"
    decoding.plugin.name = "pgoutput"
    slot.name = "seatunnel_opengauss_cdc"
  }
}

transform {

}

sink {
  jdbc {
    plugin_input = "customers_opengauss_cdc"
    url = "jdbc:postgresql://opengauss_cdc_e2e:5432/opengauss_cdc"
    driver = "org.postgresql.Driver"
    user = "dailai"
    password = "openGauss@123"

    compatible_mode="postgresLow"
    generate_sink_sql = true
    # You need to configure both database and table
    database = "opengauss_cdc"
    schema = "inventory"
    tablePrefix = "sink_"
    primary_keys = ["id"]
  }
}

支持自定义主键​

source {
  Opengauss-CDC {
    plugin_output = "customers_opengauss_cdc"
    username = "gaussdb"
    password = "openGauss@123"
    database-names = ["opengauss_cdc"]
    table-names = ["opengauss_cdc.inventory.full_types_no_primary_key"]
    url = "jdbc:postgresql://opengauss_cdc_e2e:5432/opengauss_cdc?loggerLevel=OFF"
    decoding.plugin.name = "pgoutput"
    exactly_once = true
    slot.name = "seatunnel_opengauss_cdc"
    table-names-config = [
      {
        table = "opengauss_cdc.inventory.full_types_no_primary_key"
        primaryKeys = ["id"]
      }
    ]
  }
}

CDC 元数据字段​

Opengauss CDC 会提供以下元数据字段，可配合 Metadata 转换使用：

字段类型说明
databaseSTRING源数据库名称。
tableSTRING源表名称。
rowKindSTRING变更类型，例如 insert、update 或 delete。
ts_msLONG源事件时间，单位为毫秒。
delayLONG事件时间和处理时间之间的延迟，单位为毫秒。

示例：

transform {
  Metadata {
    metadata_fields {
      Database = database
      Table = table
      RowKind = rowKind
      EventTime = ts_ms
      Delay = delay
    }
  }
}

变更日志​