离线读取节点配置
参数 | 说明 |
数据来源 | 可用的 KingbaseES 数据源。 |
库 | 支持选择、或者手动输入需读取的库名称。 默认将数据源绑定的数据库作为默认库，其他数据库需手动输入库名称。 当数据源网络不联通导致无法直接拉取库信息时，可手动输入数据库名称。在数据集成网络连通的情况下，仍可进行数据同步。 |
Schema | 支持选择、或者手动输入需读取的 Schema 名称。 |
表 | 支持选择、或者手动输入需读取的表名称。 分表情况下，可在源端支持选择或输入多个表名称，多个表需保证结构一致。 分表情况下，支持配置表序号区间。例如 table_[0-99] 表示读取 table_0、table_1、table_2 直到 table_99；如果您的表数字后缀的长度一致，例如 table_000、table_001、table_002 直到 table_999，您可以配置为 "table": ["table_00[0-9]", "table_0[10-99]", "table_[100-999]"]。 当数据源网络不联通导致无法直接拉取表信息时，可手动输入表名称。在数据集成网络连通的情况下，仍可进行数据同步。 |
添加分库分表 | 可创建多个数据源并添加对应的表对象。 注意： 分库分表情况下选择的多个表对象需保证 Schema 信息一致（包括字段名称、字段类型）。数据字段模块内系统默认展示第一个数据源的第一张表的元数据字段信息，若多表间字段不一致可能会导致运行失败。 |
切割键 | 指定用于数据分片的字段，指定后将启动并发任务进行数据同步，提升数据同步效率。您可以将源数据表中某一列作为切分键，建议使用主键或有索引的列作为切分键，仅支持类型为整型的字段。 |
筛选条件（选填） | 在实际业务场景中，通常会选择当天的数据进行同步，将 where 条件指定为 gmt_create>$bizdate。where 条件可以有效地进行业务增量同步。 如果不填写 where 语句，包括不提供 where 的 key 或 value，数据同步均视作同步全量数据。 不可以将 where 条件指定为 limit 10，这不符合 KingbaseES WHERE 子句约束。 |
高级设置（选填） | 可根据业务需求配置参数。 |
离线写入节点配置
参数 | 说明 |
数据去向 | 需要写入的 KingbaseES 数据源。 |
库 | 支持选择、或者手动输入需写入的库名称。 默认将数据源绑定的数据库作为默认库，其他数据库需手动输入库名称。 当数据源网络不联通导致无法直接拉取库信息时，可手动输入数据库名称。在数据集成网络连通的情况下，仍可进行数据同步。 |
Schema | 支持选择、或者手动输入需写入的 Schema 名称。 |
表 | 支持选择、或者手动输入需写入的表名称。 当数据源网络不联通导致无法直接拉取表信息时，可手动输入表名称。在数据集成网络连通的情况下，仍可进行数据同步。 |
是否清空表 | 在写入该 KingbaseES 数据表前可以手动选择是否清空该数据表。 |
写入模式 | KingbaseES 写入支持两种模式： append：当主键/唯一性索引冲突时，冲突行无法写入 。 upsert：主键/唯一性索引冲突时，新行会更新已指定的字段。指定的字段是指在同步任务的字段映射配置处所添加的字段。 |
批量提交大小 | 一次性批量提交的记录数大小，该值可以极大减少数据同步系统与 KingbaseES 的网络交互次数，并提升整体吞吐量。如果该值设置过大，会导致数据同步运行进程 OOM 异常。 |
前置 SQL（选填） | 执行同步任务之前执行的 SQL 语句，根据数据源类型对应的正确 SQL 语法填写 SQL，例如，执行前清空表中的旧数据（truncate table tablename）。 |
后置 SQL（选填） | 执行同步任务之后执行的 SQL 语句，根据数据源类型对应的正确 SQL 语法填写 SQL，例如，加上某一个时间戳 alter table tablename add colname timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP。 |
高级设置（选填） | 可根据业务需求配置参数。 |
数据类型转换支持
读取
KingbaseES 读取支持的字段类型及类型转换对应关系如下（在处理 KingbaseES 时，会将 KingbaseES 数据源的数据类型和数据处理引擎的数据类型做映射）：
KingbaseES 数据类型 | 内部类型 |
smallint，integer，bigint，serial，bigserial | Long |
real，double precision，numeric | Double |
varchar，char，text，money | String |
date，time，timestamp | Date |
bool | Boolean |
bytea，bit | Bytes |
写入
KingbaseES 写入支持的字段类型及类型转换对应关系如下：
内部类型 | KingbaseES 数据类型 |
Long | smallint，integer，bigint，serial，bigserial |
Double | real，double precision，numeric |
String | varchar，char，text，money |
Date | date，time，timestamp |
Boolean | bool |
Bytes | bytea，bit |
KingbaseES 脚本 Demo
{"core": {"transport": {"channel": {"speed": {"byte": -1}}}},"job": {"content": [{"reader": {"parameter": {"password": "******","column": ["id","name"],"connection": [{"jdbcUrl": ["jdbc:kingbase8://ip:54321/database"],"table": ["schema.source_table"]}],"where": "id>10","splitPk": "id","username": "kingbase"},"name": "kingbaseesreader"},"transformer": [],"writer": {"parameter": {"postSql": [""],"password": "******","session": ["set session sql_mode='ANSI'"],"column": ["id","name"],"connection": [{"jdbcUrl": "jdbc:kingbase8://ip:54321/database","table": ["schema.sink_table"]}],"writeMode": "replace","batchSize": 1024,"username": "kingbase","on": ["id"],"preSql": [""]},"name": "kingbaseeswriter"}}],"setting": {"memory": {"taskLimit": 335544320,"channelLimit": 67108864,"writerLimit": 134217728,"jobLimit": 872415232,"readerLimit": 134217728},"errorLimit": {"record": 0},"speed": {"byte": -1,"channel": 1}}}}
