配置说明

配置示例

当从Kafka同步数据到数据表或者时序表时配置项不同，且不同工作模式下相应配置文件的示例不同。此处以同步数据到数据表中为例介绍配置示例。同步数据到时序表的配置示例中需要增加时序相关配置项。

• .properties格式配置文件的示例，适用于standalone模式。

  # 设置连接器名称。
  name=tablestore-sink
  # 指定连接器类。
  connector.class=TableStoreSinkConnector
  # 设置最大任务数。
  tasks.max=1
  # 指定导出数据的Kafka的Topic列表。
  topics=test
  # 以下为Tablestore连接参数的配置。
  # Tablestore实例的Endpoint。 
  tablestore.endpoint=https://xxx.xxx.ots.aliyuncs.com
  # 填写AccessKey ID和AccessKey Secret。
  tablestore.access.key.id=xxx
  tablestore.access.key.secret=xxx
  # Tablestore实例名称。
  tablestore.instance.name=xxx
  
  # 以下为数据映射相关的配置。
  # 指定Kafka Record的解析器。
  # 默认的DefaulteEventParser已支持Struct和Map类型，您也可以使用自定义的EventParser。
  event.parse.class=com.aliyun.tablestore.kafka.connect.parsers.DefaultEventParser
  
  # 定义目标表名称的格式字符串，字符串中可包含<topic>作为原始Topic的占位符。
  # topics.assign.tables配置的优先级更高，如果配置了topics.assign.tables，则忽略table.name.format的配置。
  # 例如当设置table.name.format为kafka_<topic>时，如果kafka中主题名称为test，则将映射到Tablestore的表名为kafka_test。
  table.name.format=<topic>
  # 指定Topic与目标表的映射关系，以"<topic>:<tablename>"格式映射Topic和表名，Topic和表名之间的分隔符为英文冒号（:），不同映射之间分隔符为英文逗号（,）。
  # 如果缺省，则采取table.name.format的配置。
  # topics.assign.tables=test:test_kafka
  
  # 指定主键模式，可选值包括kafka、record_key和record_value，默认值为kafka。
  # kafka表示以<connect_topic>_<connect_partition>和<connect_offset>作为数据表的主键。
  # record_key表示以Record Key中的字段作为数据表的主键。
  # record_value表示以Record Value中的字段作为数据表的主键。
  primarykey.mode=kafka
  
  # 定义导入数据表的主键列名和数据类型。
  # 属性名格式为tablestore.<tablename>.primarykey.name和tablestore.<tablename>.primarykey.type。
  # 其中<tablename>为数据表名称的占位符。
  # 当主键模式为kafka时，无需配置该属性，默认主键列名为{"topic_partition","offset"}，默认主键列数据类型为{string, integer}。
  # 当主键模式为record_key或record_value时，必须配置以下两个属性。
  # tablestore.test.primarykey.name=A,B
  # tablestore.test.primarykey.type=string,integer
  
  # 定义属性列白名单，用于过滤Record Value中的字段获取所需属性列。
  # 默认值为空，使用Record Value中的所有字段作为数据表的属性列。
  # 属性名格式为tablestore.<tablename>.columns.whitelist.name和tablestore.<tablename>.columns.whitelist.type。
  # 其中<tablename>为数据表名称的占位符。
  # tablestore.test.columns.whitelist.name=A,B
  # tablestore.test.columns.whitelist.type=string,integer
  
  # 以下为写入Tablestore相关的配置。
  # 指定写入模式，可选值包括put和update，默认值为put。
  # put表示覆盖写。
  # update表示更新写。
  insert.mode=put
  # 是否需要保序，默认值为true。如果关闭保序模式，则有助于提升写入效率。
  insert.order.enable=true
  # 是否自动创建目标表，默认值为false。
  auto.create=false
  
  # 指定删除模式，可选值包括none、row、column和row_and_column，默认值为none。
  # none表示不允许进行任何删除。
  # row表示允许删除行。
  # column表示允许删除属性列。
  # row_and_column表示允许删除行和属性列。
  delete.mode=none
  
  # 写入数据表时内存中缓冲队列的大小，默认值为1024，单位为行数。此配置项的值必须为2的指数。
  buffer.size=1024
  # 写入数据表时的回调线程数，默认值为核数+1。
  # max.thread.count=
  # 写入数据表时的最大请求并发数，默认值为10。
  max.concurrency=10
  # 写入数据表时的分桶数，默认值为3。适当调大此配置项的值可提升并发写入能力，但不应大于最大请求并发数。
  bucket.count=3
  # 写入数据表时对缓冲区的刷新时间间隔，默认值为10000，单位为毫秒。
  flush.Interval=10000
  
  # 以下为脏数据处理相关配置。
  # 在解析Kafka Record或者写入数据表时可能发生错误，您可以可通过以下配置进行处理。
  # 指定容错能力，可选值包括none和all，默认值为none。
  # none表示任何错误都将导致Sink Task立即失败。
  # all表示跳过产生错误的Record，并记录该Record。
  runtime.error.tolerance=none
  # 指定脏数据记录模式，可选值包括ignore、kafka和tablestore，默认值为ignore。
  # ignore表示忽略所有错误。
  # kafka表示将产生错误的Record和错误信息存储在Kafka的另一个Topic中。
  # tablestore表示将产生错误的Record和错误信息存储在Tablestore另一张数据表中。
  runtime.error.mode=ignore
  
  # 当脏数据记录模式为kafka时，需要配置Kafka集群地址和Topic。
  # runtime.error.bootstrap.servers=localhost:9092
  # runtime.error.topic.name=errors
  
  # 当脏数据记录模式为tablestore时，需要配置Tablestore中数据表名称。
  # runtime.error.table.name=errors

• .json格式配置文件的示例，适用于distributed模式。

  {
    "name": "tablestore-sink",
    "config": {
      // 指定连接器类。
      "connector.class":"TableStoreSinkConnector",
      // 设置最大任务数。
      "tasks.max":"3",
      // 指定导出数据的Kafka的Topic列表。
      "topics":"test",
      // 以下为Tablestore连接参数的配置。
      // Tablestore实例的Endpoint。
      "tablestore.endpoint":"https://xxx.xxx.ots.aliyuncs.com",
      // 填写AccessKey ID和AccessKey Secret。
      "tablestore.access.key.id":"xxx",
      "tablestore.access.key.secret":"xxx",
      // Tablestore实例名称。
      "tablestore.instance.name":"xxx",
  
      // 以下为数据映射相关的配置。
      // 指定Kafka Record的解析器。
      // 默认的DefaulteEventParser已支持Struct和Map类型，您也可以使用自定义的EventParser。
      "event.parse.class":"com.aliyun.tablestore.kafka.connect.parsers.DefaultEventParser",
  
      // 定义目标表名称的格式字符串，字符串中可包含<topic>作为原始Topic的占位符。
      // topics.assign.tables配置的优先级更高。如果配置了topics.assign.tables，则忽略table.name.format的配置。
      // 例如当设置table.name.format为kafka_<topic>时，如果kafka中主题名称为test，则将映射到Tablestore的表名为kafka_test。
      "table.name.format":"<topic>",
      // 指定Topic与目标表的映射关系，以"<topic>:<tablename>"格式映射Topic和表名，Topic和表名之间的分隔符为英文冒号（:），不同映射之间分隔符为英文逗号（,）。
      // 如果缺省，则采取table.name.format的配置。
      // "topics.assign.tables":"test:test_kafka",
  
      // 指定主键模式，可选值包括kafka、record_key和record_value，默认值为kafka。
      // kafka表示以<connect_topic>_<connect_partition>和<connect_offset>作为数据表的主键。
      // record_key表示以Record Key中的字段作为数据表的主键。
      // record_value表示以Record Value中的字段作为数据表的主键。
      "primarykey.mode":"kafka",
  
      // 定义导入数据表的主键列名和数据类型。
      // 属性名格式为tablestore.<tablename>.primarykey.name和tablestore.<tablename>.primarykey.type。
      // 其中<tablename>为数据表名称的占位符。
      // 当主键模式为kafka时，无需配置该属性，默认主键列名为{"topic_partition","offset"}，默认主键列数据类型为{string, integer}。
      // 当主键模式为record_key或record_value时，必须配置以下两个属性。
      // "tablestore.test.primarykey.name":"A,B",
      // "tablestore.test.primarykey.type":"string,integer",
  
      // 定义属性列白名单，用于过滤Record Value中的字段获取所需属性列。
      // 默认值为空，使用Record Value中的所有字段作为数据表的属性列。
      // 属性名格式为tablestore.<tablename>.columns.whitelist.name和tablestore.<tablename>.columns.whitelist.type。
      // 其中<tablename>为数据表名称的占位符。
      // "tablestore.test.columns.whitelist.name":"A,B",
      // "tablestore.test.columns.whitelist.type":"string,integer",
  
      // 以下为写入Tablestore相关的配置。
      // 指定写入模式，可选值包括put和update，默认值为put。
      // put表示覆盖写。
      // update表示更新写。
      "insert.mode":"put",
      // 是否需要保序，默认值为true。如果关闭保序模式，则有助于提升写入效率。
      "insert.order.enable":"true",
      // 是否自动创建目标表，默认值为false。
      "auto.create":"false",
  
      // 指定删除模式，可选值包括none、row、column和row_and_column，默认值为none。
      // none表示不允许进行任何删除。
      // row表示允许删除行。
      // column表示允许删除属性列。
      // row_and_column表示允许删除行和属性列。
      "delete.mode":"none",
  
      // 写入数据表时内存中缓冲队列的大小，默认值为1024，单位为行数。此配置项的值必须为2的指数。
      "buffer.size":"1024",
      // 写入数据表时的回调线程数，默认值为核数+1。
      // "max.thread.count":
      // 写入数据表时的最大请求并发数，默认值为10。
      "max.concurrency":"10",
      // 写入数据表时的分桶数，默认值为3。适当调大此配置项的值可提升并发写入能力，但不应大于最大请求并发数。
      "bucket.count":"3",
      // 写入数据表时对缓冲区的刷新时间间隔，默认值为10000，单位为毫秒。
      "flush.Interval":"10000",
  
      // 以下为脏数据处理相关配置。
      // 在解析Kafka Record或者写入数据表时可能发生错误，您可以通过以下配置进行处理。
      // 指定容错能力，可选值包括none和all，默认值为none。
      // none表示任何错误都将导致Sink Task立即失败。
      // all表示跳过产生错误的Record，并记录该Record。
      "runtime.error.tolerance":"none",
      // 指定脏数据记录模式，可选值包括ignore、kafka和tablestore，默认值为ignore。
      // ignore表示忽略所有错误。
      // kafka表示将产生错误的Record和错误信息存储在Kafka的另一个Topic中。
      // tablestore表示将产生错误的Record和错误信息存储在Tablestore另一张数据表中。
      "runtime.error.mode":"ignore"
  
      // 当脏数据记录模式为kafka时，需要配置Kafka集群地址和Topic。
      // "runtime.error.bootstrap.servers":"localhost:9092",
      // "runtime.error.topic.name":"errors",
  
      // 当脏数据记录模式为tablestore时，需要配置Tablestore中数据表名称。
      // "runtime.error.table.name":"errors",
    }

配置项说明

配置文件中的配置项说明请参见下表。其中时序相关配置项只有同步数据到时序表时才需要配置。

connector.class=com.aliyun.tablestore.kafka.connect.TableStoreSinkConnector

附录：Kafka和Tablestore数据类型映射

Kafka和Tablestore数据类型映射关系请参见下表。

附录：删除语义

当同步数据到数据表且Kafka消息记录的value中存在空值时，根据写入模式（insert.mode）和删除模式（delete.mode）的不同设置，数据写入到表格存储数据表中的处理方式不同，详细说明请参见下表。