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

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

左側のナビゲーションペインで データソース をクリックし、データソースの追加 をクリックしてデータソースタイプを選択します。この方法は MySQL、PolarDB、PolarDB-O をサポートしています。

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

作成が完了すると、[データソースリスト] ページで各シャーディングデータベースの MySQL データソースを確認できます。たとえば、データソース名が MySQL_20241114 で、データベース user_info_20241114 に対応する場合です。

[シャーディングデータソース]を作成し、複数の標準データソースを 1 つのデータソースに統合して、データ同期のソースとして使用します。

1. データソースの追加 をクリックし、[シャーディングデータソース] を選択します。[データソースを追加] ダイアログボックスで、[シャーディングデータソース] タブをクリックします。利用可能なデータソースタイプには、[MySQL (シャーディング)]、[PolarDB-O (シャーディング)]、[PolarDB (シャーディング)] があります。

2. 必要に応じてシャーディングデータソースのタイプを選択し、関連パラメーターを設定します。このトピックでは MySQL (シャーディング) を例として使用します。

   [データソース名] フィールドに、Database_and_Table_Sharding_of_MySQL などの名前を入力します。設定が完了したら、[OK] をクリックしてシャーディングデータソースを作成します。

   主なパラメーターの説明：

   • データソース名 をカスタマイズします。

   • [データソースの選択]：作成済みの各シャーディングデータベースの標準データソースを選択します。

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

     重要

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

バッチ同期ノードの作成

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

ウィザードモードでシャーディングを設定します。データソース で [MySQL (シャーディング)] を選択し、作成したシャーディングデータソースを選択します。完全なタスク設定手順については、「ウィザードモードでのバッチ同期タスクの設定」をご参照ください。

ネットワーク接続テストページでは、ソースデータソース名は Database_and_Table_Sharding_of_MySQL、リソースグループは Serveless_Resource (China (Shanghai)、従量課金、0.5 CU)、宛先は MaxCompute (ODPS) で、データソース名は doc_test_project です。ネットワーク接続テストの結果、3 つすべてが接続されていることが示されます。

次へ をクリックします。

同期するテーブルを選択し、保存 と 公開 をクリックして、残りの手順を完了します。

[ソースと宛先の設定] ページのステップ 2 の [タスクの設定] では、左側の [ソース] セクションが次のように設定されます。データソースは [MySQL (シャーディング)]、テーブルの選択方法は [固定テーブル]、テーブルは [user_info]、分割キーは [id] です。(シャーディングでは、すべてのテーブルが同一のスキーマを持つ必要があります。そうでない場合、実行エラーが発生します。) 右側の [宛先] セクションは次のように設定されます。データソースは MaxCompute (ODPS)、テーブルは [user_info]、パーティション情報は pt=${bizdate}、書き込みモードは、書き込み前に既存データをクリア (Insert Overwrite)、空の文字列を NULL に変換する設定は [いいえ] です。

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

バッチ同期ノードの作成

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

ウィザードモードでシャーディングを設定します。データソース セクションで、[+ ソースを編集] をクリックします。表示されるデータソース選択パネルで、複数のシャーディングデータベースに対応するデータソースを選択して追加します。完全なタスク設定手順については、「ウィザードモードでのバッチ同期タスクの設定」をご参照ください。

選択済みのデータソース に複数の標準データソースを追加し、OK をクリックします。

次へ をクリックします。

各データソースから同期するテーブルを選択し、保存 と 公開 をクリックして、残りの手順を完了します。

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

バッチ同期ノードの作成

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

スクリプトモードでシャーディングを設定します。以下にスクリプトの例を示します。完全なタスク設定手順については、「スクリプトモードでのバッチ同期タスクの設定」をご参照ください。

{
    "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"
                }
        ]
    }
}