Tablestore:Data Integration

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

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

• RAMユーザーが作成され、Tablestore管理権限（AliyunOTSFullAccess）とDataWorks管理権限（AliyunDataWorksFullAccess）が付与されている。詳細については、RAMユーザーを作成するとRAMユーザーに権限を付与するを参照してください。

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

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

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

  重要

  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":"age","type":"INTEGER"}アイコンをクリックします。これはに対応します。ソースフィールドを表すソースフィールドの名前を指定できます。デスティネーションフィールドを指定するには、JSON形式を使用する必要があります。例：。各行はフィールドを表します。

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

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

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

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

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

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

   チャネル制御 セクションで、同期レートや不正なデータレコードのポリシーなどのタスク実行のためのパラメーターを設定します。詳細については、手順 5: チャネル制御ポリシーの設定を参照してください。

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

   説明

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

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

   重要

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

   1. をクリックします。

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

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

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

      現在のタスクステータスの値がFINISHの場合は、タスクは完了です。

Tablestoreコンソールにログオンしてデータを表示します。

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

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

3. データが復元されるTablestoreインスタンスの名前をクリックします。

4. インスタンスの詳細タブのテーブルタブで、復元されたテーブルの名前をクリックします。

5. データのクエリタブで、Tablestoreにインポートされたデータを表示できます。