Tablestore:構成の説明

構成例

構成項目は、Kafka から Tablestore のデータテーブルまたは時系列テーブルにデータを同期するかどうかに応じて異なります。構成ファイルの構成例は、動作モードによって異なります。このセクションでは、Kafka から Tablestore のデータテーブルへのデータ同期を構成する方法の例を示します。Tablestore の時系列テーブルにデータを同期するには、Kafka から Tablestore の時系列テーブルへのデータ同期に固有の構成項目を追加する必要があります。

• 次のサンプルコードは、スタンドアロンモードで Tablestore Sink Connector の .properties 形式の構成ファイルを構成する方法の例を示しています。

  # コネクタ名を指定します。
  name=tablestore-sink
  # コネクタクラスを指定します。
  connector.class=TableStoreSinkConnector
  # タスクの最大数を指定します。
  tasks.max=1
  # データのエクスポート元の Kafka トピックのリストを指定します。
  topics=test
  # 次の Tablestore 接続パラメータの値を指定します。
  # Tablestore インスタンスのエンドポイント。
  tablestore.endpoint=https://xxx.xxx.ots.aliyuncs.com
  # AccessKey ID と AccessKey シークレットで構成される AccessKey ペア。
  tablestore.access.key.id=xxx
  tablestore.access.key.secret=xxx
  # Tablestore インスタンスの名前。
  tablestore.instance.name=xxx
  
  # 次のデータマッピングパラメータを指定します。
  # Kafka メッセージレコードの解析に使用するパーサーを指定します。
  # Tablestore Sink Connector の DefaultEventParser は、Kafka Connect の Struct クラスと Map クラスをサポートしています。カスタム EventParser を使用することもできます。
  event.parse.class=com.aliyun.tablestore.kafka.connect.parsers.DefaultEventParser
  
  # エクスポート元のトピックのプレースホルダーとして <topic> を使用して、宛先 Tablestore テーブルの名前のフォーマット文字列を指定します。
  # topics.assign.tables は table.name.format よりも優先されます。topics.assign.tables が指定されている場合、table.name.format の構成は無視されます。
  # たとえば、table.name.format が kafka_<topic> に設定され、データのエクスポート元の Kafka トピックの名前が test の場合、test トピックからの Kafka メッセージレコードは Tablestore の kafka_test という名前のテーブルにマッピングされます。
  table.name.format=<topic>
  # Kafka トピックと宛先 Tablestore テーブル間のマッピングを指定します。値は <topic>:<tablename> 形式である必要があります。トピック名とテーブル名はコロン（:）で区切ります。複数のマッピングを指定する場合は、カンマ（,）で区切ります。
  # マッピングが指定されていない場合は、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_value：レコード値のフィールドがデータテーブルのプライマリキーとして使用されます。
  primarykey.mode=kafka
  
  # 宛先 Tablestore データテーブルのプライマリキー列の名前とデータ型を指定します。
  # プライマリキー列名の形式は 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
  
  # 属性列のホワイトリストを指定して、レコード値のフィールドをフィルタリングし、必要な属性列を取得します。
  # デフォルトでは、属性列のホワイトリストは空です。レコード値のすべてのフィールドがデータテーブルの属性列として使用されます。
  # 属性列名の形式は 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
  
  # Kafka メッセージレコードを宛先 Tablestore テーブルに書き込む方法を指定します。
  # 書き込みモードを指定します。有効な値：put と update。デフォルト値：put。
  # put：宛先テーブルのデータは Kafka メッセージレコードによって上書きされます。
  # update：宛先テーブルのデータは Kafka メッセージレコードによって更新されます。
  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
  # データをデータテーブルに書き込むときに使用されるコールバックスレッドの数を指定します。デフォルト値 = vCPU の数 + 1。
  # max.thread.count=
  # データをデータテーブルに書き込むために送信できる同時書き込みリクエストの最大数を指定します。デフォルト値：10。
  max.concurrency=10
  # データの書き込み先となるバケットの数を指定します。デフォルト値：3。このパラメータの値を増やすと、同時書き込み機能を向上させることができます。ただし、このパラメータの値を指定した同時書き込みリクエストの最大数よりも大きい値に設定することはできません。
  bucket.count=3
  # データをデータテーブルに書き込むときにバッファキューを更新する間隔を指定します。単位：ミリ秒。デフォルト値：10000。
  flush.Interval=10000
  
  # ダーティデータの処理方法を指定します。
  # Kafka メッセージレコードの解析時またはデータテーブルへの書き込み時にエラーが発生する可能性があります。次の 2 つのパラメータを指定して、エラーの修正方法を決定できます。
  # フォールトトレランス機能を指定します。有効な値：none と all。デフォルト値：none。
  # none：エラーが発生すると、Tablestore Sink Connector を使用するデータインポートタスクが失敗します。
  # all：エラーが報告されたメッセージレコードはスキップされ、ログに記録されます。
  runtime.error.tolerance=none
  # ダーティデータのログ記録方法を指定します。有効な値：ignore、kafka、tablestore。デフォルト値：ignore。
  # ignore：すべてのエラーが無視されます。
  # kafka：エラーが報告されたメッセージレコードとエラーメッセージは別の Kafka トピックに保存されます。
  # tablestore：エラーが報告されたメッセージレコードとエラーメッセージは別の Tablestore データテーブルに保存されます。
  runtime.error.mode=ignore
  
  # runtime.error.mode を kafka に設定した場合、Kafka クラスタアドレスとトピックを指定する必要があります。
  # runtime.error.bootstrap.servers=localhost:9092
  # runtime.error.topic.name=errors
  
  # runtime.error.mode を tablestore に設定した場合、Tablestore データテーブルの名前を指定する必要があります。
  # runtime.error.table.name=errors

• 次のサンプルコードは、分散モードで Tablestore Sink Connector の .json 形式の構成ファイルを構成する方法の例を示しています。

  {
    "name": "tablestore-sink",
    "config": {
      // コネクタクラスを指定します。
      "connector.class":"TableStoreSinkConnector",
      // タスクの最大数を指定します。
      "tasks.max":"3",
      // データのエクスポート元の Kafka トピックのリストを指定します。
      "topics":"test",
      // 次の Tablestore 接続パラメータの値を指定します。
      // Tablestore インスタンスのエンドポイント。
      "tablestore.endpoint":"https://xxx.xxx.ots.aliyuncs.com",
      // AccessKey ID と AccessKey シークレットで構成される AccessKey ペア。
      "tablestore.access.key.id":"xxx",
      "tablestore.access.key.secret":"xxx",
      // Tablestore インスタンスの名前。
      "tablestore.instance.name":"xxx",
  
      // 次のデータマッピングパラメータを指定します。
      // Kafka メッセージレコードの解析に使用するパーサーを指定します。
      // Tablestore Sink Connector の DefaultEventParser は、Kafka Connect の Struct クラスと Map クラスをサポートしています。カスタム EventParser を使用することもできます。
      "event.parse.class":"com.aliyun.tablestore.kafka.connect.parsers.DefaultEventParser",
  
      // エクスポート元のトピックのプレースホルダーとして <topic> を使用して、宛先 Tablestore テーブルの名前のフォーマット文字列を指定します。
      // topics.assign.tables は table.name.format よりも優先されます。topics.assign.tables が指定されている場合、table.name.format の構成は無視されます。
      // たとえば、table.name.format が kafka_<topic> に設定され、データのエクスポート元の Kafka トピックの名前が test の場合、test トピックからの Kafka メッセージレコードは Tablestore の kafka_test という名前のテーブルにマッピングされます。
      "table.name.format":"<topic>",
      // Kafka トピックと宛先 Tablestore テーブル間のマッピングを指定します。値は <topic>:<tablename> 形式である必要があります。トピック名とテーブル名はコロン（:）で区切ります。複数のマッピングを指定する場合は、カンマ（,）で区切ります。
      // マッピングが指定されていない場合は、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_value：レコード値のフィールドがデータテーブルのプライマリキーとして使用されます。
      "primarykey.mode":"kafka",
  
      // 宛先 Tablestore データテーブルのプライマリキー列の名前とデータ型を指定します。
      // プライマリキー列名の形式は 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",
  
      // 属性列のホワイトリストを指定して、レコード値のフィールドをフィルタリングし、必要な属性列を取得します。
      // デフォルトでは、属性列のホワイトリストは空です。レコード値のすべてのフィールドがデータテーブルの属性列として使用されます。
      // 属性列名の形式は 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",
  
      // Kafka メッセージレコードを宛先 Tablestore テーブルに書き込む方法を指定します。
      // 書き込みモードを指定します。有効な値：put と update。デフォルト値：put。
      // put：テーブルのデータは Kafka メッセージレコードによって上書きされます。
      // update：テーブルのデータは Kafka メッセージレコードによって更新されます。
      "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",
      // データをデータテーブルに書き込むときに使用されるコールバックスレッドの数を指定します。デフォルト値 = vCPU の数 + 1。
      // "max.thread.count":
      // データをデータテーブルに書き込むために送信できる同時書き込みリクエストの最大数を指定します。デフォルト値：10。
      "max.concurrency":"10",
      // データの書き込み先となるバケットの数を指定します。デフォルト値：3。このパラメータの値を増やすと、同時書き込み機能を向上させることができます。ただし、このパラメータの値を指定した同時書き込みリクエストの最大数よりも大きい値に設定することはできません。
      "bucket.count":"3",
      // データをデータテーブルに書き込むときにバッファキューを更新する間隔を指定します。単位：ミリ秒。デフォルト値：10000。
      "flush.Interval":"10000",
  
      // ダーティデータの処理方法を指定します。
      // Kafka メッセージレコードの解析時またはデータテーブルへの書き込み時にエラーが発生する可能性があります。次の 2 つのパラメータを指定して、エラーの修正方法を決定できます。
      // フォールトトレランス機能を指定します。有効な値：none と all。デフォルト値：none。
      // none：エラーが発生すると、Tablestore Sink Connector を使用するデータインポートタスクが失敗します。
      // all：エラーが報告されたメッセージレコードはスキップされ、ログに記録されます。
      "runtime.error.tolerance":"none",
      // ダーティデータのログ記録方法を指定します。有効な値：ignore、kafka、tablestore。デフォルト値：ignore。
      // ignore：すべてのエラーが無視されます。
      // kafka：エラーが報告されたメッセージレコードとエラーメッセージは別の Kafka トピックに保存されます。
      // tablestore：エラーが報告されたメッセージレコードとエラーメッセージは別の Tablestore データテーブルに保存されます。
      "runtime.error.mode":"ignore"
  
      // runtime.error.mode を kafka に設定した場合、Kafka クラスタアドレスとトピックを指定する必要があります。
      // "runtime.error.bootstrap.servers":"localhost:9092",
      // "runtime.error.topic.name":"errors",
  
      // runtime.error.mode を tablestore に設定した場合、Tablestore データテーブルの名前を指定する必要があります。
      // "runtime.error.table.name":"errors",
    }

パラメータ

次の表は、構成ファイルのパラメータについて説明しています。Kafka から Tablestore の時系列テーブルにデータを同期する場合にのみ、時系列関連のパラメータを構成する必要があります。

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

付録：Kafka と Tablestore 間のデータ型マッピング

次の表は、Kafka と Tablestore のデータ型間のマッピングについて説明しています。

付録：削除構文

次の表は、メッセージレコードに空の値が含まれており、Kafka から Tablestore のデータテーブルにデータが同期されている場合に、書き込みモード (insert.mode) と削除モード (delete.mode) の構成に基づいて Tablestore データテーブルにデータを書き込むために使用されるメソッドについて説明しています。