Skip to main content

KingBase Sink

一、介绍

KingBase Sink支持将数据写入到Kingbase指定的数据库中。

二、支持版本

KingBase v8

三、插件名称

类型名称
Synckingbasesink、kingbasewriter
SQLkingbase-x

四、参数说明

1、Sync

  • connection

    • 描述:数据库连接参数,包含jdbcUrl、schema、table等参数
    • 必选:是
    • 参数类型:List
    • 默认值:无
         "connection": [{
      "jdbcUrl":["jdbc:kingbase8://localhost:54321/MOWEN"],
      "table": ["table"],
      "schema":"public"
      }]

  • jdbcUrl

    • 描述:针对关系型数据库的jdbc连接字符串,jdbcUrl参考: Kingbase帮助文档
    • 必选:是
    • 参数类型:string
    • 默认值:无
  • schema

    • 描述:数据库schema名
    • 必选:否
    • 参数类型:string
    • 默认值:无
  • table

    • 描述:目的表的表名称。目前只支持配置单个表,后续会支持多表
    • 必选:是
    • 参数类型:List
    • 默认值:无
  • username

    • 描述:数据源的用户名
    • 必选:是
    • 参数类型:String
    • 默认值:无
  • password

    • 描述:数据源指定用户名的密码
    • 必选:是
    • 参数类型:String
    • 默认值:无
  • column

    • 描述:目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: "column": ["id","name","age"]
    • 必选:是
    • 参数类型:List
    • 默认值:无
  • fullcolumn

    • 描述:目的表中的所有字段,字段之间用英文逗号分隔。例如: "column": ["id","name","age","hobby"],如果不配置,将在系统表中获取
    • 必选:否
    • 参数类型:List
    • 默认值:无
  • preSql

    • 描述:写入数据到目的表前,会先执行这里的一组标准语句
    • 必选:否
    • 参数类型:List
    • 默认值:无
  • postSql

    • 描述:写入数据到目的表后,会执行这里的一组标准语句
    • 必选:否
    • 参数类型:List
    • 默认值:无
  • writeMode

    • 描述:控制写入数据到目标表采用 insert into 或者 replace into 或者 ON DUPLICATE KEY UPDATE 语句
    • 必选:是
    • 所有选项:insert/replace/update
    • 参数类型:String
    • 默认值:insert
  • batchSize

    • 描述:一次性批量提交的记录数大小,该值可以极大减少FlinkX与数据库的网络交互次数,并提升整体吞吐量。但是该值设置过大可能会造成FlinkX运行进程OOM情况
    • 必选:否
    • 参数类型:int
    • 默认值:1024
  • updateKey

    • 描述:当写入模式为update和replace时,需要指定此参数的值为唯一索引字段
    • 注意:
      • 如果此参数为空,并且写入模式为update和replace时,应用会自动获取数据库中的唯一索引;
      • 如果数据表没有唯一索引,但是写入模式配置为update和replace,应用会以insert的方式写入数据;
    • 必选:否
    • 参数类型:Map<String,List>
      • 示例:"updateKey": {"key": ["id"]}
    • 默认值:无
  • semantic

    • 描述:sink端是否支持二阶段提交
    • 注意:
      • 如果此参数为空,默认不开启二阶段提交,即sink端不支持exactly_once语义;
      • 当前只支持exactly-once 和at-least-once
    • 必选:否
    • 参数类型:String
      • 示例:"semantic": "exactly-once"
    • 默认值:at-least-once

2、SQL

  • connector

    • 描述:kingbase-x
    • 必选:是
    • 参数类型:String
    • 默认值:无
  • url

    • 描述:jdbc:kingbase8://localhost:54321/MOWEN
    • 必选:是
    • 参数类型:String
    • 默认值:无
  • schema

    • 描述:数据库schema名
    • 必选:否
    • 参数类型:string
    • 默认值:无
  • table-name

    • 描述:表名
    • 必选:是
    • 参数类型:String
    • 默认值:无:
  • username

    • 描述:username
    • 必选:是
    • 参数类型:String
    • 默认值:无
  • password

    • 描述:password
    • 必选:是
    • 参数类型:String
    • 默认值:无
  • sink.buffer-flush.max-rows

    • 描述:批量写数据条数,单位:条
    • 必选:否
    • 参数类型:String
    • 默认值:1024
  • sink.buffer-flush.interval

    • 描述:批量写时间间隔,单位:毫秒
    • 必选:否
    • 参数类型:String
    • 默认值:10000
  • sink.all-replace

    • 描述:是否全部替换数据库中的数据(如果数据库中原值不为null,新值为null,如果为true则会替换为null)
    • 必选:否
    • 参数类型:String
    • 默认值:false
  • sink.parallelism

    • 描述:写入结果的并行度
    • 必选:否
    • 参数类型:String
    • 默认值:无
  • sink.semantic

    • 描述:sink端是否支持二阶段提交
    • 注意:
      • 如果此参数为空,默认不开启二阶段提交,即sink端不支持exactly_once语义;
      • 当前只支持exactly-once 和at-least-once
    • 必选:否
    • 参数类型:String
      • 示例:"semantic": "exactly-once"
    • 默认值:at-least-once

五、数据类型

是否支持类型
支持SMALLINT、SMALLSERIAL、INT2、INT、INTEGER、SERIAL、INT4、BIGINT、BIGSERIAL、OID、INT8、REAL、FLOAT4、FLOAT、DOUBLE PRECISION、FLOAT8、DECIMAL、NUMERIC、 CHARACTER VARYING、VARCHAR、CHARACTER、CHAR、TEXT、NAME、BPCHAR、BYTEA、TIMESTAMP、TIMESTAMPTZ、DATE、TIME、TIMETZ、 BOOLEAN、BOOL
暂不支持ARRAY等

六、脚本示例

见项目内flinkx-examples文件夹。