Tablestore：配置說明

配置樣本

當從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）的不同設定，資料寫入到Tablestore資料表中的處理方式不同，詳細說明請參見下表。