Tablestore:増分データを OSS に同期する

(推奨) コードレス UI を使用する

1. タスクを設定する ステップの ソースとデスティネーションを設定する セクションで、ビジネス要件に基づいてソースデータソースとデスティネーションデータソースを設定します。

   ソースデータソースを設定する

   パラメーター	説明
   テーブル	Tablestore のデータテーブルの名前。
   開始時刻	増分データが読み取られる期間の開始時刻と終了時刻。${startTime} および ${endTime} 形式でパラメーターを指定します。具体的な形式は、スケジューリングプロパティで設定されます。期間は左閉右開区間です。
   終了時刻
   状態テーブル	Tablestore Stream の状態を格納するテーブルの名前。デフォルト値: TableStoreStreamReaderStatusTable。
   最大再試行回数	Tablestore から増分データを読み取る各リクエストで許可される最大再試行回数。
   時系列情報をエクスポートする	時系列情報をエクスポートするかどうかを指定します。時系列情報には、データが書き込まれた時刻が含まれます。

   デスティネーションデータソースを設定する

   パラメーター	説明
   テキストタイプ	OSS に書き込まれるオブジェクトの形式 (.csv、.txt など)。 説明 オブジェクトの形式が異なると、必要な設定も異なります。オブジェクトの形式を選択し、表示される特定のパラメーターを指定します。
   ファイル名 (パスを含む)	このパラメーターは、テキストタイプ パラメーターを text、csv、または orc に設定した場合にのみ表示されます。 OSS 内のオブジェクトの名前。名前には、tablestore/20231130/myotsdata.csv などのパスを含めることができます。
   パス	このパラメーターは、テキストタイプ パラメーターを parquet に設定した場合にのみ表示されます。 OSS 内のオブジェクトのパス。例: tablestore/20231130/。
   ファイル名	このパラメーターは、テキストタイプ パラメーターを parquet に設定した場合にのみ表示されます。 OSS 内のオブジェクトの名前。
   列区切り文字	このパラメーターは、テキストタイプ パラメーターを text または csv に設定した場合にのみ表示されます。 データが OSS オブジェクトに書き込まれるときに列を区切るために使用する区切り文字。
   行区切り文字	このパラメーターは、テキストタイプ パラメーターを text に設定した場合にのみ表示されます。 行を区切るために使用するカスタム区切り文字。例: \u0001。データに存在しない区切り文字を行区切り文字として使用する必要があります。 Linux のデフォルトの行区切り文字 \n または Windows の \r\n を使用する場合、このパラメーターを空のままにすることをお勧めします。システムは、デフォルトの行区切り文字に基づいてデータを自動的に読み取ります。
   コーディング	このパラメーターは、テキストタイプ パラメーターを text または csv に設定した場合にのみ表示されます。 データを書き込む OSS オブジェクトのエンコーディング形式。
   null 値	このパラメーターは、テキストタイプ パラメーターを text、csv、または orc に設定した場合にのみ表示されます。 ソースデータソースで null として解釈できる文字列。たとえば、このパラメーターを null に設定し、ソースデータソースのフィールドの値が null の場合、システムはそのフィールドを null として認識します。
   時刻形式	このパラメーターは、テキストタイプ パラメーターを text または csv に設定した場合にのみ表示されます。 日付データが OSS オブジェクトに書き込まれる時刻形式。例: yyyy-MM-dd。
   プレフィックスの競合	指定したオブジェクト名が OSS 内の既存のオブジェクトの名前と同じである場合に使用される処理方法。有効な値: 置換: 既存のオブジェクトを削除し、同じ名前の別のオブジェクトを作成します。 保持: 既存のオブジェクトを保持し、既存のオブジェクト名とランダムなサフィックスを組み合わせた名前の別のオブジェクトを作成します。 エラー: エラーを報告し、バッチ同期タスクを停止します。
   ファイルを分割する	このパラメーターは、テキストタイプ パラメーターを text または csv に設定した場合にのみ表示されます。 データの書き込み先となる単一の OSS オブジェクトの最大サイズ。単位: MB。最大サイズは 100 GB です。書き込まれるオブジェクトのサイズがこのパラメーターの値に達すると、システムは新しいオブジェクトを生成してデータの書き込みを続行し、すべてのデータが書き込まれるまで続けます。
   単一ファイルとして書き込む	このパラメーターは、テキストタイプ パラメーターを text または csv に設定した場合にのみ表示されます。 一度に 1 つのオブジェクトを OSS に書き込むかどうかを指定します。デフォルトでは、一度に複数のオブジェクトが OSS に書き込まれます。データの書き込み中にデータが読み取られない場合、次のいずれかのケースが発生する可能性があります。オブジェクトヘッダーが設定されている場合、オブジェクトヘッダーのみを含む空のオブジェクトが生成されます。オブジェクトヘッダーが設定されていない場合、完全に空のオブジェクトが生成されます。 一度に 1 つのオブジェクトを OSS に書き込む場合は、単一ファイルとして書き込む を選択します。この場合、データの書き込み中にデータが読み取られないと、空のファイルは生成されません。
   最初の行をテーブルヘッダーにする	このパラメーターは、テキストタイプ パラメーターを text または csv に設定した場合にのみ表示されます。 データがオブジェクトに書き込まれるときに最初の行をテーブルヘッダーとして書き込むかどうかを指定します。デフォルトでは、テーブルヘッダーは生成されません。最初の行をテーブルヘッダーとして書き込む場合は、最初の行をテーブルヘッダーにする を選択します。

2. フィールドマッピング セクションでは、システムがフィールドマッピングを自動的に完了します。このセクションでは、デフォルト設定を保持します。

   ソースフィールドには、プライマリキー列とデータテーブルの増分データの変更が含まれます。デスティネーションフィールドは設定できません。

   

3. チャネル制御 セクションで、タスクの予想最大同時実行数、同期レート、ダーティデータレコードのポリシー、分散実行パラメーターなど、タスク実行のパラメーターを設定します。パラメーターの詳細については、「チャネル制御ポリシーを設定する」を参照してください。

4. アイコンをクリックして、設定を保存します。

   説明

   設定を保存しないと、後続の操作を実行したときに設定を保存するように促すメッセージが表示されます。OK をクリックして設定を保存します。

コードエディターを使用する

増分データを同期するには、OTSStream Reader と OSS Writer を使用する必要があります。コードエディターを使用してタスクを設定する方法の詳細については、「Tablestore Stream データソース」と「OSS データソース」を参照してください。

1. タスクを設定する ステップで、 アイコンをクリックします。表示されるメッセージで、OK をクリックします。

2. コードエディターで、次のサンプルコードに基づいてパラメーターを指定します。

   重要

   設定を理解しやすいように、サンプルコードにコメントが記載されています。サンプルコードを使用する場合は、すべてのコメントを削除してください。

   行モードに基づいて増分データをエクスポートする

   {
       "type": "job",
       "version": "2.0",
       "steps": [
           {
               "stepType": "otsstream", // リーダーの名前。このパラメーターの値は変更できません。
               "parameter": {
                   "statusTable": "TableStoreStreamReaderStatusTable", // Tablestore Stream の状態を格納するテーブルの名前。ほとんどの場合、このパラメーターの値を変更する必要はありません。
                   "maxRetries": 30, // 各リクエストで許可される最大再試行回数。
                   "isExportSequenceInfo": false, // 時系列情報をエクスポートするかどうかを指定します。時系列情報には、データが書き込まれた時刻が含まれます。
   				        "mode": "single_version_and_update_only", // Tablestore Stream を使用してデータをエクスポートするモード。このパラメーターを single_version_and_update_only に設定します。設定テンプレートにこのパラメーターが含まれていない場合は、このパラメーターを追加します。
                   "datasource": "otssource", // ソースデータソースの名前。ビジネス要件に基づいてこのパラメーターを指定します。
                   "envType": 1,
                   "column": [ 
   				          {
                        "name": "pk1" 
                     },
                     {
                        "name": "pk2"  
                     },
                     {
                        "name": "col1"  
                     }
                   ],
                   "startTimeString": "${startTime}", // データエクスポートの開始時刻。このタスクは増分データエクスポートに使用されるため、ループで開始する必要があります。各ループの開始時刻は異なります。したがって、${startTime} などの変数を使用する必要があります。
                   "table": "mytable", // Tablestore のソーステーブルの名前。
                   "endTimeString": "${endTime}" // データエクスポートの終了時刻。${endTime} などの変数を使用する必要があります。
               },
               "name": "Reader",
               "category": "reader"
           },
           {
               "stepType": "oss", // ライターの名前。このパラメーターの値は変更できません。
               "parameter": { // このサンプルコードに基づいてパラメーターを指定する場合、データは CSV 形式と TEXT 形式でのみエクスポートできます。Parquet 形式または ORC 形式でデータをエクスポートする場合は、「OSS データソース」トピックの内容に基づいてパラメーターを指定します。
                   "fieldDelimiterOrigin": ",",
                   "nullFormat": "null", // null フィールド値を識別するために使用される文字列。値は空の文字列にすることができます。
                   "dateFormat": "yyyy-MM-dd HH:mm:ss", // 時刻の形式。
                   "datasource": "osssource", // OSS データソースの名前。ビジネス要件に基づいてこのパラメーターを指定します。
                   "envType": 1,
                   "writeSingleObject": true, // 一度に 1 つのファイルを OSS に書き込むかどうかを指定します。true の値は、一度に 1 つのファイルを OSS に書き込むことを指定します。データが読み取られない場合、空のファイルは生成されません。false の値は、一度に複数のファイルを OSS に書き込むことを指定します。データが読み取られず、ファイルヘッダーが設定されている場合、ファイルヘッダーのみを含む空のファイルが生成されます。それ以外の場合、空のファイルが生成されます。
                   "writeMode": "truncate", // デスティネーションデータソースに指定されたソースファイル名と同じ名前のオブジェクトが存在する場合にシステムによって実行される操作。有効な値: truncate、append、nonConflict。truncate の値は、システムがデスティネーションデータソースのオブジェクトをクリアすることを指定します。append の値は、システムがデスティネーションデータソースのオブジェクトにデータを追加することを指定します。nonConflict の値は、エラーが報告されることを指定します。
                   "encoding": "UTF-8", // エンコーディングタイプ。
                   "fieldDelimiter": ",", // 列を区切るために使用する区切り文字。
                   "fileFormat": "csv", // ファイル形式。有効な値: csv、text、parquet、orc。
                   "object": "" // OSS に同期するファイルの名前のプレフィックス。「Tablestore インスタンス名/テーブル名/日付」形式を使用することをお勧めします。例: "instance/table/{date}"。
               },
               "name": "Writer",
               "category": "writer" 
           },
           {
               "copies": 1,
               "parameter": {
                   "nodes": [],
                   "edges": [],
                   "groups": [],
                   "version": "2.0"
               },
               "name": "Processor",
               "category": "processor"
           }
       ],
       "setting": {
           "errorLimit": {
               "record": "0" // 許可されるエラーの数。エラーの数がこの値を超えると、同期タスクは失敗します。
           },
           "locale": "zh",
           "speed": {
               "throttle": false, // スロットリングを有効にするかどうかを指定します。false の値はスロットリングが無効になっていることを指定し、true の値はスロットリングが有効になっていることを指定します。mbps パラメーターは、throttle パラメーターが true に設定されている場合にのみ有効になります。
               "concurrent": 3 // 同時タスクの最大数。
           }
       },
       "order": {
           "hops": [
               {
                   "from": "Reader",
                   "to": "Writer"
               }
           ]
       }
   }

   行の列値の増分変更に基づいて増分データをエクスポートする

   {
       "type": "job",
       "version": "2.0",
       "steps": [
           {
               "stepType": "otsstream", // リーダーの名前。このパラメーターの値は変更できません。
               "parameter": {
                   "statusTable": "TableStoreStreamReaderStatusTable", // Tablestore Stream の状態を格納するテーブルの名前。ほとんどの場合、このパラメーターの値を変更する必要はありません。
                   "maxRetries": 30, // 各リクエストで許可される最大再試行回数。
                   "isExportSequenceInfo": false, // 時系列情報をエクスポートするかどうかを指定します。時系列情報には、データが書き込まれた時刻が含まれます。
                   "datasource": "otssource",
                   "envType": 1,
                   "column": [ // データテーブルから OSS にエクスポートする列。設定テンプレートにこのパラメーターが含まれていない場合は、このパラメーターを追加します。この例では、デフォルト設定が使用されています。
                       "pk1", // プライマリキー列の名前。データテーブルに複数のプライマリキー列がある場合は、すべて指定する必要があります。
                       "pk2", // プライマリキー列の名前。データテーブルに複数のプライマリキー列がある場合は、すべて指定する必要があります。
                       "colName", // 値が増分データである属性列の名前。このパラメーターの値を変更する必要はありません。
                       "version", // 値が増分データである列のデータバージョン番号。このパラメーターの値を変更する必要はありません。このパラメーターの値は 64 ビットのタイムスタンプです。単位: ミリ秒。
                       "colValue", // 値が増分データである属性列の値。このパラメーターの値を変更する必要はありません。
                       "opType", // 増分データ操作のタイプ。このパラメーターの値を変更する必要はありません。
                       "sequenceInfo" // 自動インクリメントシーケンスの ID。このパラメーターの値を変更する必要はありません。
                   ],
                   "startTimeString": "${startTime}", // データエクスポートの開始時刻。このタスクは増分データエクスポートに使用されるため、ループで開始する必要があります。各ループの開始時刻は異なります。したがって、${startTime} などの変数を使用する必要があります。
                   "table": "mytable", // Tablestore のソーステーブルの名前。
                   "endTimeString": "${endTime}" // データエクスポートの終了時刻。${endTime} などの変数を使用する必要があります。
               },
               "name": "Reader",
               "category": "reader"
           },
           {
               "stepType": "oss", // ライターの名前。このパラメーターの値は変更できません。
               "parameter": { // このサンプルコードに基づいてパラメーターを指定する場合、データは CSV 形式と TEXT 形式でのみエクスポートできます。Parquet 形式または ORC 形式でデータをエクスポートする場合は、「OSS データソース」トピックの内容に基づいてパラメーターを指定します。
                   "fieldDelimiterOrigin": ",",
                   "nullFormat": "null", // null フィールド値を識別するために使用される文字列。値は空の文字列にすることができます。
                   "dateFormat": "yyyy-MM-dd HH:mm:ss", // 時刻の形式。
                   "datasource": "osssource", // OSS データソースの名前。ビジネス要件に基づいてこのパラメーターを指定します。
                   "envType": 1,
                   "writeSingleObject": true, // 一度に 1 つのファイルを OSS に書き込むかどうかを指定します。true の値は、一度に 1 つのファイルを OSS に書き込むことを指定します。データが読み取られない場合、空のファイルは生成されません。false の値は、一度に複数のファイルを OSS に書き込むことを指定します。データが読み取られず、ファイルヘッダーが設定されている場合、ファイルヘッダーのみを含む空のファイルが生成されます。それ以外の場合、空のファイルが生成されます。
                   "column": [ // OSS に同期する列。列のシーケンス番号を使用してこのパラメーターを指定します。このパラメーターの値を変更する必要はありません。
                       "0",
                       "1",
                       "2",
                       "3",
                       "4",
                       "5",
                       "6"
                   ],
                   "writeMode": "truncate", // デスティネーションデータソースに指定されたソースファイル名と同じ名前のオブジェクトが存在する場合にシステムによって実行される操作。有効な値: truncate、append、nonConflict。truncate の値は、システムがデスティネーションデータソースのオブジェクトをクリアすることを指定します。append の値は、システムがデスティネーションデータソースのオブジェクトにデータを追加することを指定します。nonConflict の値は、エラーが報告されることを指定します。
                   "encoding": "UTF-8", // エンコーディングタイプ。
                   "fieldDelimiter": ",", // 列を区切るために使用する区切り文字。
                   "fileFormat": "csv", // ファイル形式。有効な値: csv、text。
                   "object": "" // OSS に同期するファイルの名前のプレフィックス。「Tablestore インスタンス名/テーブル名/日付」形式を使用することをお勧めします。例: "instance/table/{date}"。
               },
               "name": "Writer",
               "category": "writer"
           },
           {
               "copies": 1,
               "parameter": {
                   "nodes": [],
                   "edges": [],
                   "groups": [],
                   "version": "2.0"
               },
               "name": "Processor",
               "category": "processor"
           }
       ],
       "setting": {
           "errorLimit": {
               "record": "0" // 許可されるエラーの数。エラーの数がこの値を超えると、同期タスクは失敗します。
           },
           "locale": "zh",
           "speed": {
               "throttle": false, // スロットリングを有効にするかどうかを指定します。false の値はスロットリングが無効になっていることを指定し、true の値はスロットリングが有効になっていることを指定します。mbps パラメーターは、throttle パラメーターが true に設定されている場合にのみ有効になります。
               "concurrent": 3 // 同時タスクの最大数。
           }
       },
       "order": {
           "hops": [
               {
                   "from": "Reader",
                   "to": "Writer"
               }
           ]
       }
   }

3. アイコンをクリックして、設定を保存します。

   説明

   スクリプトを保存しないと、後続の操作を実行したときにスクリプトを保存するように促すメッセージが表示されます。OK をクリックしてスクリプトを保存します。