Tablestore:Data Integration

後続の操作を実行する前に、以下の準備が整っていることを確認してください。

• Tablestore テーブルに同期する MaxCompute テーブルの情報が確認され、記録されている。

• RAM ユーザーが作成され、Tablestore 管理権限 (AliyunOTSFullAccess) と DataWorks 管理権限 (AliyunDataWorksFullAccess) が付与されている。 詳細については、「RAM ユーザーの作成」と「RAM ユーザーへの権限付与」をご参照ください。

• RAM ユーザーの AccessKey ペアが作成されている。 詳細については、「AccessKey ペアを作成する」をご参照ください。

• DataWorks がアクティブ化され、ワークスペースが作成されている。 詳細については、「DataWorks をアクティブ化する」と「ワークスペースを作成する」をご参照ください。

• Tablestore がアクティブ化されている。 インスタンスとデータテーブルが作成されている。 詳細については、「Tablestore をアクティブ化してインスタンスを作成する」と「Tablestore をアクティブ化する」をご参照ください。

  重要

  Tablestore データテーブルのプライマリキー列の名前と型は、MaxCompute テーブルのプライマリキー列の名前と型と一致する必要があります。 Tablestore データテーブルに属性列を定義する必要はありません。 Tablestore データテーブルにデータを書き込むときに、属性列の名前を指定できます。

この例では、Tablestore データソースを追加します。 ビジネス要件に基づいて必要なデータソースを追加してください。

1. Data Integration ページに移動します。

   DataWorks コンソール にログインし、左上隅でリージョンを選択し、[データ開発とガバナンス] > [data Integration] を選択し、ドロップダウンリストからワークスペースを選択して、[data Integration に移動] をクリックします。

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

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

4. [データソースの追加] ダイアログボックスで、[tablestore] ブロックをクリックします。

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

   パラメータ	説明
   データソース名	データソースの名前。 名前に使用できるのは、英字、数字、アンダースコア (_) で、最初の文字は英字にする必要があります。
   データソースの説明	データソースの説明。 説明は 80 文字以内にする必要があります。
   エンドポイント	Tablestore インスタンスのエンドポイント。 詳細については、「エンドポイント」をご参照ください。 Tablestore インスタンスとデスティネーションデータソースのリソースが同じリージョンにある場合は、VPC エンドポイントを入力します。 そうでない場合は、パブリックエンドポイントを入力します。
   Table Store インスタンス名	Tablestore インスタンスの名前。 詳細については、「インスタンス」をご参照ください。
   AccessKey ID	Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID と AccessKey シークレット。 AccessKey ペアの作成方法については、「AccessKey ペアを作成する」をご参照ください。
   AccessKey Secret

6. データソースと選択したリソースグループ間のネットワーク接続をテストします。

   同期ノードが想定どおりに実行されるようにするには、データソースと、同期ノードが実行されるすべてのリソースグループタイプ間の接続をテストする必要があります。

   重要

   同期タスクは、1 つのタイプのリソースグループのみを使用できます。 デフォルトでは、Data Integration の共有リソースグループのみがリソースグループリストに表示されます。 データ同期の安定性とパフォーマンスを確保するために、Data Integration 専用リソースグループを使用することをお勧めします。

   1. [購入] をクリックして新しいリソースグループを作成するか、[購入したリソースグループを関連付ける] をクリックして既存のリソースグループを関連付けます。 詳細については、「Data Integration 専用リソースグループを作成および使用する」をご参照ください。

   2. リソースグループが開始されたら、リソースグループの [接続ステータス (本番環境)] 列の [ネットワーク接続のテスト] をクリックします。

      [接続済み] と表示されている場合は、接続テストに合格しています。

7. データソースがネットワーク接続テストに合格した場合は、[完了] をクリックします。

   新しく作成されたデータソースがデータソースリストに表示されます。

1. 同期タスクノードを追加します。

   1. [datastudio] コンソールに移動します。

      DataWorks コンソール にログインし、左上隅でリージョンを選択し、左側のナビゲーションウィンドウで [データ開発とガバナンス] > [データ開発] を選択し、ドロップダウンリストからワークスペースを選択して、[datastudio に移動] をクリックします。

   2. DataStudio コンソールの [スケジュールされたワークフロー] セクションで、[ビジネスフロー] をクリックしてビジネスフローを選択します。

      ワークフローの作成方法については、「ワークフローを作成する」をご参照ください。

   3. [data Integration] を右クリックし、[ノードの作成] > [オフライン同期] を選択します。

   4. [ノードの作成] ダイアログボックスで、[名前] フィールドに名前を入力し、[確認] をクリックします。

      作成したバッチ同期ノードは、[data Integration] の下にあります。

2. ネットワークとリソースのリンクを構成します。

   バッチ同期タスクのソースとデスティネーション、およびバッチ同期タスクの実行に使用するリソースグループを選択します。 リソースグループとデータソース間のネットワーク接続を確立し、接続をテストします。

   1. [data Integration] の下で、作成したバッチ同期ノードをダブルクリックします。

   2. [ネットワーク接続とリソースグループの構成] 手順で、[ソース] パラメータを [maxcompute(odps)] に設定し、[データソース名] パラメータを MaxCompute データソースの名前に設定します。

   3. リソースグループを選択します。

      リソースグループを選択すると、リソースグループのリージョンと仕様が表示され、リソースグループと選択したデータソース間の接続が自動的にテストされます。

      重要

      リソースグループが、データソースを追加したときに選択したリソースグループと同じであることを確認してください。

   4. [デスティネーション] パラメータを [tablestore] に設定し、[データソース名] パラメータを Tablestore データソースの名前に設定します。

      リソースグループとデスティネーション間の接続が自動的にテストされます。

   5. 接続テストに合格したら、[次へ] をクリックします。

3. ソースとデスティネーションを構成します。

   1. [ソースとデスティネーションの構成] セクションで、ソースのパラメータを構成します。 次の表に、パラメータを示します。

      パラメータ	説明
      トンネルリソースグループ	MaxCompute のデータ伝送リソース。 デフォルト値は [共通伝送リソース] で、これは MaxCompute の無料クォータです。
      テーブル	MaxCompute のテーブルの名前。
      パーティション情報	データを読み取るパーティションの名前。

   2. [ソースとデスティネーションの構成] セクションで、デスティネーションのパラメータを構成します。 次の表に、パラメータを示します。

      パラメータ	説明
      テーブル	Tablestore データテーブルの名前。
      primaryKeyInfo	Tablestore データテーブルに関するプライマリキー情報。 このパラメータの値は JSON 配列です。 データテーブルのプライマリキーは、1 ～ 4 つのプライマリキー列で構成されます。 プライマリキー列の実際の数に基づいてプライマリキー情報を指定します。 重要 指定するプライマリキー列の数、順序、型は、データテーブルのプライマリキー列の数、順序、型と同じである必要があります。 例: [ { "name": "pk", "type": "string" } ]
      WriteMode	Tablestore にデータを書き込むモード。 有効値: PutRow: 指定した行にデータを挿入します。 このモードは、Tablestore PutRow API 操作に対応しています。 指定した行が存在しない場合は、新しい行が追加されます。 指定した行が存在する場合は、その行が上書きされます。 UpdateRow: 指定した行のデータを更新します。 このモードは、Tablestore UpdateRow API 操作に対応しています。 指定した行が存在しない場合は、新しい行が追加されます。 指定した行が存在する場合は、リクエストの内容に基づいて、行内の指定した列の値が追加、変更、または削除されます。

4. フィールドマッピングを構成します。

   ソースとデスティネーションを構成したら、ソースフィールドとデスティネーションフィールド間のマッピングを構成する必要があります。 マッピングを構成すると、バッチ同期タスクは、マッピングに基づいてソースフィールドの値を特定のデータ型のデスティネーションフィールドに書き込みます。 デフォルトでは、ソースの行のフィールドは、デスティネーションの同じ行のフィールドにマップされます。

   重要

   • マッピング順序を変更する場合は、ターゲット フィールド[ソースフィールド] に対応する {"name":"年齢","type":"整数"} の横にある アイコンをクリックします。 ソースフィールドを表すソースフィールドの名前を指定できます。 デスティネーションフィールドを指定するには、JSON 形式を使用する必要があります。 例: 。 各行はフィールドを表します。

   • Tablestore データテーブルの属性列のデータ型は、MaxCompute に書き込むフィールドのデータ型と一致する必要があります。 Tablestore データテーブルの属性列の名前は、MaxCompute に書き込むフィールドの名前と異なっていても構いません。

   デフォルトでは、デスティネーションのプライマリキー情報は、デスティネーションのパラメータを構成するときに指定されます。 この場合、デスティネーションフィールドのプライマリキー情報を手動で削除する必要があります。 そうしないと、バッチ同期タスクは失敗します。

   1. アイコンの横にある [ターゲット フィールド] をクリックします。

   2. [テーブルフィールドの手動編集] ダイアログボックスで、デスティネーションフィールドのプライマリキー情報を削除し、ビジネス要件に基づいて属性列情報を追加または変更します。

   3. [確認] をクリックします。

5. チャネル制御ポリシーを構成します。

   [チャンネル コントロール] セクションで、同期レートやダーティ データ レコードのポリシーなどのタスク実行のためのパラメーターを構成します。詳細については、「手順 5: チャンネル コントロール ポリシーを構成する」をご参照ください。

6. アイコンをクリックして構成を保存します。

   説明

   構成を保存しないと、後続の操作を実行するときに構成を保存するように求めるメッセージが表示されます。 この場合は、メッセージの [確認] をクリックして構成を保存します。

7. 同期タスクを実行します。

   重要

   ほとんどの場合、完全データを同期する必要があるのは 1 回だけです。 スケジュールプロパティを構成する必要はありません。

   1. アイコンをクリックします。

   2. [パラメータ] ダイアログボックスで、ドロップダウンリストからリソースグループの名前を選択します。

   3. [実行] をクリックします。

      同期タスクが完了したら、[ランタイムログ] タブの実行ログの URL をクリックして、実行ログの詳細ページに移動します。 実行ログの詳細ページで、Current task status の値を確認します。

      Current task status の値が FINISH の場合、タスクは完了です。

Tablestore コンソールにログインして、データ同期の結果を表示します。

1. Tablestore コンソール にログインします。

2. [概要] ページの上部で、リージョンを選択します。

3. データの同期先である Tablestore インスタンスの名前をクリックします。

4. [インスタンスの詳細] タブの [テーブル] タブで、データテーブルの名前をクリックします。

5. [データのクエリ] タブで、データテーブルに同期されたデータを表示します。