DataWorks:Tablestore ストリームデータソース

データ同期前に Tablestore ストリーム環境を準備する

SyncClient client = new SyncClient("", "", "", "");
// 方法 1: テーブルを作成するときに Stream 機能を有効にします。
CreateTableRequest createTableRequest = new CreateTableRequest(tableMeta);
createTableRequest.setStreamSpecification(new StreamSpecification(true, 24)); // 値 24 は、Tablestore が増分データを 24 時間保持することを示します。
client.createTable(createTableRequest);
// 方法 2: テーブルを作成するときに Stream 機能を有効にしない場合は、テーブルの作成後に UpdateTable オペレーションを呼び出して、この機能を有効にすることができます。
UpdateTableRequest updateTableRequest = new UpdateTableRequest("tableName");
updateTableRequest.setStreamSpecification(new StreamSpecification(true, 24));
client.updateTable(updateTableRequest);

サポートされている読み取りモードとデータ型

Tablestore Stream Reader を使用すると、Tablestore から列モードまたは行モードで増分データを読み取ることができます。このセクションでは、読み取りモードのデータ読み取りプロセスとサポートされているデータ型について説明します。

列モードでデータを読み取る

Tablestore のマルチバージョンモードでは、テーブルデータは 3 レベル構造 で編成されます。行、列、バージョンです。1 つの行に複数の列を含めることができ、列名は固定されていません。各列に複数のバージョンを含めることができ、各バージョンには特定のタイムスタンプ（バージョン番号）があります。

Tablestore の API オペレーションを呼び出して、データの読み取りと書き込みを行うことができます。Tablestore は、テーブルデータに対して実行された最近の書き込みおよび変更操作を記録して、増分データを記録します。したがって、増分データは一連の操作レコードと見なすことができます。

Tablestore は動的列とマルチバージョンモードをサポートしています。したがって、Tablestore Stream Reader によって生成される行は、Tablestore の行ではなく、列のバージョンです。Tablestore Stream Reader が Tablestore の行からデータを読み取った後、Tablestore Stream Reader はデータを複数の行に変換します。各行には、プライマリキー値、列名、列のバージョン（バージョン番号）のタイムスタンプ、バージョンの値、および操作タイプが含まれます。isExportSequenceInfo パラメーターが true に設定されている場合は、時系列情報も含まれます。

行モードでデータを読み取る

• ワイドテーブル

  Tablestore Stream Reader を使用して、行モードでデータを読み取ることもできます。このモードでは、Tablestore Stream Reader は操作レコードを行として読み取ります。mode パラメーターを設定し、データを読み取る列を指定する必要があります。

  "parameter": {
    // mode を single_version_and_update_only に設定し、isExportSequenceInfo を false に設定し、ビジネス要件に基づいて datasource や table などの他のパラメーターを設定します。
    "mode": "single_version_and_update_only", // 読み取りモード。
    "column":[  // Tablestore からデータを読み取る列。ビジネス要件に基づいて列を指定できます。
            {
               "name": "uid"  // 列の名前。プライマリキー列またはプロパティ列を指定できます。
            },
            {
               "name": "name"  // 列の名前。プライマリキー列またはプロパティ列を指定できます。
            },
    ],
    "isExportSequenceInfo": false, // 時系列情報を読み取るかどかを指定します。mode パラメーターを single_version_and_update_only に設定した場合、このパラメーターは false にのみ設定できます。
  }

• 時系列テーブル

  時系列テーブルを作成すると、Stream 機能は自動的に有効になります。

  Tablestore Stream Reader を使用すると、時系列テーブルから増分データを読み取ることができます。テーブルが時系列テーブルの場合、次のパラメーターを設定する必要があります。

  "parameter": {
    // ビジネス要件に基づいて、次のパラメーターと datasource や table などの他のパラメーターを設定します。
    "mode": "single_version_and_update_only", // 読み取りモード。
    "isTimeseriesTable":"true", // テーブルが時系列テーブルかどうかを指定します。
    "column":[  // Tablestore からデータを読み取る列。ビジネス要件に基づいて列を指定できます。
            {
              "name": "_m_name" // メトリック名列の名前。
            },
            {
              "name": "_data_source" // データソース列の名前。
            },
            {
              "name": "_tags" // タグ列の名前。タグは文字列型のデータに変換されます。
            },
            {
              "name": "tag1", // タグキーの名前。
              "is_timeseries_tag":"true"  // フィールドがタグの内部フィールドかどうかを指定します。
            },
            {
              "name": "time" // タイムスタンプ列の名前。
            },
            {
               "name": "name" // プロパティ列の名前。
            },
    ],
    "isExportSequenceInfo": false, // 時系列情報を読み取るかどかを指定します。mode パラメーターを single_version_and_update_only に設定した場合、このパラメーターは false にのみ設定できます。
  }

  行ごとに読み取られるデータは、元の行のデータに近くなります。これにより、データのさらなる処理が容易になります。行ごとにデータを読み取る場合は、次の点に注意してください。

  • 読み取られる行は、操作レコードから抽出されます。各行は、書き込みまたは更新操作に対応しています。行の一部の列のみを更新する場合、操作レコードには更新された列のみが含まれます。

  • 各列のバージョン番号（各列のタイムスタンプ）は、読み取ったり削除したりすることはできません。

データ型のマッピング

データ同期タスクの開発: Tablestore からの増分同期のプロシージャガイダンス

• 設定手順については、「コードレス UI を使用してバッチ同期タスクを設定する」および「コードエディターを使用してバッチ同期タスクを設定する」をご参照ください。

• コードエディターを使用してバッチ同期タスクを設定する場合に設定されるすべてのパラメーターと実行されるコードについては、「付録: コードとパラメーター」をご参照ください。

付録: コードとパラメーター

付録: コードエディターを使用してバッチ同期タスクを設定する

コードエディターを使用してバッチ同期タスクを設定する場合は、コードエディターの形式要件に基づいて、関連データソースのリーダーとライターのパラメーターを設定する必要があります。形式要件の詳細については、「コードエディターを使用してバッチ同期タスクを設定する」をご参照ください。次の情報では、コードエディターのリーダーとライターのパラメーターの設定の詳細について説明します。

Tablestore Stream Reader のコード

• 列モードでデータを読み取る

  {
      "type":"job",
      "version":"2.0",// バージョン番号。
      "steps":[
          {
              "stepType":"otsstream",// プラグイン名。
              "parameter":{
                  "datasource":"$srcDatasource",// データソースの名前。
                  "dataTable":"",// テーブルの名前。
                  "statusTable":"TableStoreStreamReaderStatusTable",// Tablestore Stream Reader がステータスレコードの保存に使用するテーブルの名前。
                  "maxRetries":30,// Tablestore から増分データを読み取る各リクエストの最大再試行回数。デフォルト値: 30。
                  "isExportSequenceInfo":false,// 時系列情報を読み取るかどかを指定します。
                  "startTimeString":"${startTime}${hh}",// 増分データの開始時刻。同期タスクの実行開始時刻と同じです。このパラメーターは yyyymmddhh24miss 形式で設定します。
                  "endTimeString":"${endTime}${hh}"// 増分データの終了時刻。同期タスクの実行完了時刻と同じです。このパラメーターは yyyymmddhh24miss 形式で設定します。
              },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"// 許容されるダーティデータレコードの最大数。
          },
          "speed":{
              "throttle":true,// スロットリングを有効にするかどうかを指定します。値 false はスロットリングが無効になっていることを示し、値 true はスロットリングが有効になっていることを示します。mbps パラメーターは、throttle パラメーターが true に設定されている場合にのみ有効になります。
              "concurrent":1,// 並列スレッドの最大数。
              "mbps":"12"// 最大伝送速度。単位: MB/秒。
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

• ワイドテーブルから行モードでデータを読み取る

  {
      "type":"job",
      "version":"2.0",// バージョン番号。
      "steps":[
          {
              "stepType":"otsstream",// プラグイン名。
              "parameter":{
                  "datasource":"$srcDatasource",// データソースの名前。
                  "dataTable":"",// テーブルの名前。
                  "statusTable":"TableStoreStreamReaderStatusTable",// Tablestore Stream Reader がステータスレコードの保存に使用するテーブルの名前。
                  "maxRetries":30,// Tablestore から増分データを読み取る各リクエストの最大再試行回数。デフォルト値: 30。
                  "isExportSequenceInfo":false,// 時系列情報を読み取るかどかを指定します。
                  "startTimeString":"${startTime}${hh}",// 増分データの開始時刻。同期タスクの実行開始時刻と同じです。このパラメーターは yyyymmddhh24miss 形式で設定します。
                  "endTimeString":"${endTime}${hh}"// 増分データの終了時刻。同期タスクの実行完了時刻と同じです。このパラメーターは yyyymmddhh24miss 形式で設定します。
                  "mode": "single_version_and_update_only",
                  "column":[
                          {
                              "name":"pId"
                          },
                          {
                              "name": "uId"
                          },
                          {
                              "name":"col0"
                          },
                          {
                              "name": "col1"
                          }
                      ],
                      },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"// 許容されるダーティデータレコードの最大数。
          },
          "speed":{
              "throttle":true,// スロットリングを有効にするかどうかを指定します。値 false はスロットリングが無効になっていることを示し、値 true はスロットリングが有効になっていることを示します。mbps パラメーターは、throttle パラメーターが true に設定されている場合にのみ有効になります。
              "concurrent":1,// 並列スレッドの最大数。
              "mbps":"12" // 最大伝送速度。
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

• 時系列テーブルから行モードでデータを読み取る

  {
      "type":"job",
      "version":"2.0",// バージョン番号。
      "steps":[
          {
              "stepType":"otsstream",// プラグイン名。
              "parameter":{
                  "datasource":"$srcDatasource",// データソースの名前。
                  "dataTable":"",// テーブルの名前。
                  "statusTable":"TableStoreStreamReaderStatusTable",// Tablestore Stream Reader がステータスレコードの保存に使用するテーブルの名前。
                  "maxRetries":30,// Tablestore から増分データを読み取る各リクエストの最大再試行回数。デフォルト値: 30。
                  "isExportSequenceInfo":false,// 時系列情報を読み取るかどかを指定します。
                  "startTimeString":"${startTime}${hh}",// 増分データの開始時刻。同期タスクの実行開始時刻と同じです。このパラメーターは yyyymmddhh24miss 形式で設定します。
                  "endTimeString":"${endTime}${hh}"// 増分データの終了時刻。同期タスクの実行完了時刻と同じです。このパラメーターは yyyymmddhh24miss 形式で設定します。
                  "mode": "single_version_and_update_only",
                  "isTimeseriesTable":"true",
                  "column": [
                            {
                              "name": "_m_name"
                            },
                            {
                              "name": "_data_source",
                            },
                            {
                              "name": "_tags",
                            },
                            {
                              "name": "string_column",
                            }
                      ]
                      },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"// 許容されるダーティデータレコードの最大数。
          },
          "speed":{
              "throttle":true,// スロットリングを有効にするかどうかを指定します。値 false はスロットリングが無効になっていることを示し、値 true はスロットリングが有効になっていることを示します。mbps パラメーターは、throttle パラメーターが true に設定されている場合にのみ有効になります。
              "concurrent":1,// 並列スレッドの最大数。
              "mbps":"12"// 最大伝送速度。単位: MB/秒。
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

Tablestore Stream Reader のコードのパラメーター

"column":[
    {"name":"pk1"},
	{"name":"col1"},
	{"name":"col2","dataType":"new"},
	{"name":"col2","dataType":"old"},
	{"name":"col2","dataType":"latest"}
],