Tablestore:OSS への増分データの同期

ステップ 1: Tablestore データソースの追加

DataWorks で Tablestore データソースを設定して、ソースデータに接続します。

1. DataWorks コンソールにログインします。ターゲットリージョンに切り替えます。左側のナビゲーションウィンドウで、[データ統合] > [データ統合] を選択します。ドロップダウンリストからワークスペースを選択し、[データ統合へ] をクリックします。

2. 左側のナビゲーションウィンドウで、[データソース] をクリックします。

3. [データソース] ページで、[データソースの追加] をクリックします。

4. [データソースの追加] ダイアログボックスで、データソースタイプとして [Tablestore] を検索して選択します。

5. [OTS データソースの追加] ダイアログボックスで、次の表に従ってデータソースパラメーターを設定します。

   パラメーター	説明
   データソース名	データソース名は、文字、数字、アンダースコア (_) の組み合わせである必要があります。数字またはアンダースコア (_) で始めることはできません。
   データソースの説明	データソースの簡単な説明。説明は 80 文字以内にする必要があります。
   リージョン	Tablestore インスタンスが存在するリージョンを選択します。
   Tablestore インスタンス名	Tablestore インスタンスの名前。
   エンドポイント	Tablestore インスタンスのエンドポイント。VPC アドレスを使用します。
   AccessKey ID	Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID と AccessKey Secret。
   AccessKey Secret

6. リソースグループの接続性をテストします。

   データソースを作成するときは、リソースグループの接続性をテストして、同期タスクのリソースグループがデータソースに接続できることを確認する必要があります。そうしないと、データ同期タスクを実行できません。

   1. [接続設定] セクションで、リソースグループの [接続ステータス] 列にある [ネットワーク接続のテスト] をクリックします。

   2. 接続性テストに合格したら、[完了] をクリックします。新しいデータソースがデータソースリストに表示されます。

      接続性テストが失敗した場合は、[ネットワーク接続診断ツール] を使用して問題をトラブルシューティングします。

ステップ 2: OSS データソースの追加

データエクスポートの宛先として OSS データソースを設定します。

1. 再度 [データソースの追加] をクリックします。ダイアログボックスで、データソースタイプとして [OSS] を検索して選択し、データソースパラメーターを設定します。

   パラメーター	説明
   データソース名	データソース名は、文字、数字、アンダースコア (_) で構成する必要があります。数字またはアンダースコア (_) で始めることはできません。
   データソースの説明	データソースの簡単な説明。説明は 80 文字以内にする必要があります。
   アクセスモード	[RAM ロール認証モード]: DataWorks サービスアカウントは、RAM ロールを引き受けることでデータソースにアクセスします。このモードを初めて選択する場合は、画面の指示に従って必要な権限を付与します。 [AccessKey モード]: Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID と AccessKey Secret を使用してデータソースにアクセスします。
   ロール	このパラメーターは、[アクセスモード] を [RAM ロール認証モード] に設定した場合にのみ必要です。
   AccessKey ID	これらのパラメーターは、[アクセスモード] を [AccessKey モード] に設定した場合にのみ必要です。Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID と AccessKey Secret。
   AccessKey Secret
   リージョン	バケットが配置されているリージョン。
   エンドポイント	OSS ドメイン名。詳細については、「リージョンとエンドポイント」をご参照ください。
   バケット	バケットの名前。

2. パラメーターを設定し、接続性テストに合格したら、[完了] をクリックしてデータソースを追加します。

ステップ 3: バッチ同期タスクの設定

データ同期タスクを作成および設定して、Tablestore から OSS へのデータ転送ルールを定義します。

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "otsstream",
            "parameter": {
                "statusTable": "TableStoreStreamReaderStatusTable",
                "maxRetries": 31,
                "isExportSequenceInfo": false,
                "datasource": "source_data",
                "column": [
                    "id",
                    "name",
                    "colName",
                    "version",
                    "colValue",
                    "opType",
                    "sequenceInfo"
                ],
                "startTimeString": "${startTime}",
                "table": "source_table",
                "endTimeString": "${endTime}"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "oss",
            "parameter": {
                "dateFormat": "yyyy-MM-dd HH:mm:ss",
                "datasource": "target_data",
                "writeSingleObject": false,
                "column": [
                    "0",
                    "1",
                    "2",
                    "3",
                    "4",
                    "5",
                    "6"
                ],
                "writeMode": "truncate",
                "encoding": "UTF-8",
                "fieldDelimiter": ",",
                "fileFormat": "csv",
                "object": "tablestore/source_table.csv"
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "concurrent": 2,
            "throttle": false
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "otsstream",
            "parameter": {
                "statusTable": "TableStoreStreamReaderStatusTable",
                "maxRetries": 31,
                "isExportSequenceInfo": false,
                "datasource": "source_data",
                "column": [
                  {
                    "name": "_m_name"
                  },
                  {
                    "name": "_data_source"
                  },
                  {
                    "name": "_tags"
                  },
                  {
                    "name": "_time"
                  },
                  {
                    "name": "value",
                    "type": "int"
                  }
                ],
                "startTimeString": "${startTime}",
                "table": "source_series",
                "isTimeseriesTable":"true",
                "mode": "single_version_and_update_only",
                "endTimeString": "${endTime}"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "oss",
            "parameter": {
                "dateFormat": "yyyy-MM-dd HH:mm:ss",
                "datasource": "target_data",
                "writeSingleObject": false,
                "writeMode": "truncate",
                "encoding": "UTF-8",
                "fieldDelimiter": ",",
                "fileFormat": "csv",
                "object": "tablestore/source_series.csv"
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "concurrent": 2,
            "throttle": false
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

ステップ 4: 同期結果の表示

同期タスクが完了したら、ログで実行ステータスを表示し、OSS バケットで結果ファイルを確認できます。

1. ページの下部でタスクの実行ステータスと結果を表示します。次のログ情報は、同期タスクが正常に実行されたことを示しています。

   2025-11-18 11:16:23 INFO Shell run successfully!
   2025-11-18 11:16:23 INFO Current task status: FINISH
   2025-11-18 11:16:23 INFO Cost time is: 77.208s

2. 宛先バケット内のファイルを表示します。

   バケットリストに移動します。宛先バケットをクリックして、結果ファイルを表示またはダウンロードします。