DataWorks:シャーディングバッチ同期タスクの設定

シャーディングデータソースとコードレス UI

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

   DataWorks コンソールにログインします。 左側のナビゲーションウィンドウで、[Data Integration] をクリックします。 表示されるページで、ドロップダウンリストから目的のワークスペースを選択し、[Data Integrationに移動] をクリックします。

2. 左側のナビゲーションペインで、Data Source をクリックし、次に Add Data Source をクリックしてデータソースタイプを選択します。この方法では現在、MySQL、PolarDB、および PolarDB-O をサポートしています。

3. シャーディングデータベースごとに標準データソースを作成します。本トピックでは MySQL を例として使用します。詳細な手順については、「データソースの設定」をご参照ください。

   データソースの作成後、[データソースリスト] ページで各シャーディングデータベースの MySQL データソースを確認できます。例えば、MySQL_20241114 という名前のデータソースは、データベース user_info_20241114 に対応します。

4. [シャードデータベースとテーブル] データソースを作成すると、複数の標準データソースを単一のデータソースに結合して、同期元として使用できます。

   1. Add Data Source をクリックし、[シャードデータベースとテーブル] を選択します。[データソースの追加] ダイアログボックスで、[シャードデータベースとテーブル] タブをクリックします。利用可能なデータソースタイプには、[MySQL (シャードデータベースとテーブル)]、[PolarDB-O (シャードデータベースとテーブル)]、および [PolarDB (シャードデータベースとテーブル)] があります。

   2. シャーディングデータソースタイプを選択し、パラメーターを設定します。本トピックでは MySQL (シャーディングデータベースとテーブル) を例として使用します。

      [データソース名] に Database_and_Table_Sharding_of_MySQL などの名前を入力します。パラメーターの設定後、[完了] をクリックしてシャーディングデータソースを作成します。

      主なパラメーター：

      • [Data Source Name]：データソースにカスタム名を指定します。

      • [データソースの選択]: シャードデータベース用に作成した標準データソースを選択します。

      • [メタデータソース]：シャーディング同期では、すべてのデータソース内のテーブルのスキーマが同一である必要があります。デフォルトのメタデータのテンプレートとして、1 つのデータソースを選択します。このテンプレートは、同期タスクの設定時にデフォルトのスキーマを取得するために使用されます。

        重要

        データソース間でテーブルスキーマが一致していない場合、同期タスクは失敗します。

5. バッチ同期ノードを作成します。

6. シャーディング同期タスクを設定します。

   コードレス UI でシャーディングを設定するには、Data Source で [MySQL (シャーディングされたデータベースとテーブル)] を選択し、次に作成したシャーディングされたデータソースを選択します。 詳細については、「コードレス UI を使用してバッチ同期タスクを設定する」をご参照ください。

   説明

   本トピックでは MySQL (シャーディングデータベースとテーブル) を例として使用しています。業務要件に基づいてタイプを選択してください。

   ネットワーク接続テストページには、ソースデータソース名が [Database_and_Table_Sharding_of_MySQL]、リソースグループが [Serverless_Resource] (中国 (上海) 、従量課金、0.5 CU 消費) 、宛先が [MaxCompute (ODPS)] タイプのデータソース [doc_test_project] と表示されます。3 つすべての接続テストに成功しました。

7. Next をクリックします。

8. 同期するテーブルを選択し、Save と Publish をクリックし、残りの手順を完了します。

   [ソースと宛先の設定] ページの [タスクの設定] ステップで、左側の [ソース] セクションで、データソースとして [MySQL (シャーディングデータベースとテーブル)] を選択します。[テーブル選択方法] を [固定テーブル] に設定し、[user_info] テーブルを選択し、[分割キー] に [id] を入力します。（シャーディングでは、すべてのテーブルが同一のスキーマを持つ必要があります。そうでない場合、タスクは失敗します。） 右側の [宛先] セクションで、データソースとして [MaxCompute(ODPS)] を選択し、テーブル名を [user_info] に設定します。[パーティション情報] を pt=${bizdate} に設定します。[書き込みモード] を [Insert Overwrite] に設定し、[空の文字列をnullに変換] を [いいえ] に設定します。

標準データソースとコードレス UI

1. シャーディングデータベースごとに標準データソースを作成します。本トピックでは MySQL を例として使用します。詳細な手順については、「データソースの設定」をご参照ください。

2. バッチ同期ノードを作成します。

3. シャーディング同期タスクを設定します。

   コードフリー UI でシャーディングを設定するには、Data Source セクションで [+ データソースの編集] をクリックします。 表示されるデータソース選択パネルで、シャーディングされたデータベースのデータソースを選択して追加します。 詳細については、「コードフリー UI を使用してバッチ同期タスクを設定する」をご参照ください。

4. 標準データソースをSelected Data Sourcesリストに追加し、Confirmをクリックします。

5. Next をクリックします。

6. データソースごとに同期するテーブルを選択し、次にSaveとPublishをクリックして、残りの手順を完了します。

   重要

   この設定では、Data Sources for Sharded Databases Use Same Account and Password オプションがデフォルトで選択されています。これは、すべてのシャードデータソースが最初のデータソースのアカウントとパスワードを使用して、それぞれのデータベースにアクセスすることを意味します。シャードデータベースで異なるアカウントとパスワードを使用する場合は、このオプションの選択を解除します。これにより、各データソースは独自に設定された認証情報を使用できるようになります。

   [ソースと宛先の設定] ページの左側の [ソース] セクションには、MySQL_20241112、MySQL_20241114、MySQL_20241113 などの 3 つの MySQL シャーディングデータソースが設定されています。各データソースに対して、[user_info] テーブルが選択されています。ページには「シャーディングでは、すべてのテーブルが同一のスキーマを持つ必要があります。そうでない場合、タスクは失敗します。」という警告が表示されます。右側の [宛先] セクションでは、データソースが [MaxCompute (ODPS)]、テーブルが [user_info] です。[パーティション情報] は pt=${bizdate} に設定され、[書き込みモード] は [Insert Overwrite] です。

コードエディタ

1. バッチ同期ノードを作成します。

2. シャーディング同期タスクを設定します。

   コードエディタでシャーディングを設定するには、次の例のようなスクリプトを使用します。タスクの設定方法の詳細については、「コードエディタを使用したバッチ同期タスクの設定」をご参照ください。

   重要

   実行する前に、次のコードからコメントを削除してください。

   {
       "type":"job",
       "version":"2.0",
       "steps":[
           {
               "stepType":"mysql",
               "parameter":{
                   "envType":0,
                   "column":[
                       "id",
                       "name"
                   ],
                   "socketTimeout":3600000,
                   "tableComment":"",
                   "connection":[    // シャーディングデータベースの数に基づいて接続を設定します。
                       {
                           "datasource":"datasourceName1",  // シャーディングデータソース1。
                           "table":[           // データソース1のテーブル。
                               "tb1"
                           ]
                       },
                       {
                           "datasource":"datasourceName2", // シャーディングデータソース2。
                           "table":[          // データソース2のテーブル。
                               "tb2",
                               "tb3"
                           ]
                       }
                   ],
                   "useSpecialSecret":true, // 各データソースは独自のパスワードを使用します。
                   "where":"",
                   "splitPk":"id",
                   "encoding":"UTF-8"
                   },
               "name":"Reader",
               "category":"reader"
               },
           {
               "stepType":"odps",
               "parameter":{
                   "partition":"pt=${bizdate}",
                   "truncate":true,
                   "datasource":"odpsname",
                   "envType":0,
                   "isSupportThreeModel":false,
                   "column":[
                       "id",
                       "name"
                   ],
                   "emptyAsNull":false,
                   "tableComment":"",
                   "table":"t1",
                   "consistencyCommit":false
                   },
               "name":"Writer",
               "category":"writer"
               }
       ],
       "setting":{
           "executeMode":null,
           "errorLimit":{
               "record":""
               },
           "speed":{
               "concurrent":2,
               "throttle":false
               }
       },
       "order":{
           "hops":[
               {
                   "from":"Reader",
                   "to":"Writer"
                   }
           ]
       }
   }

よくある質問