DataWorks:MySQL から Elasticsearch へのリアルタイム同期

アクション

説明

Refresh

システムは、選択したソーステーブルを自動的にリストアップします。ただし、送信先インデックスのプロパティは、リフレッシュして確認した後にのみ適用されます。

• 同期する複数のテーブルを選択し、[マッピング結果を一括リフレッシュ] をクリックします。

• 送信先インデックス名：送信先インデックス名は、Customize Mapping Rules for Destination Table Names ルールに基づいて自動的に生成されます。デフォルト名は ${source_database_name}_${table_name} です。この名前のインデックスが送信先に存在しない場合、システムは自動的に作成します。

Custom Mapping Rule for Destination Index Name (任意)

システムはデフォルトのルールを使用してインデックス名を生成します：${source_database_name}_${table_name}。また、Customize Mapping Rules for Destination Table Names 列の Edit ボタンをクリックしてカスタムルールを追加することもできます。

• ルール名：ルールの名前を定義します。ビジネス目的を反映した、わかりやすい名前を指定することを推奨します。

• 送信先インデックス名： ボタンをクリックし、Manual Input と Built-in Variable の値を組み合わせて送信先インデックス名を構築できます。サポートされている変数には、ソースデータソース名、ソースデータベース名、ソーステーブル名が含まれます。

• 組み込み変数の編集：組み込み変数に文字列変換を適用できます。

この機能は、以下のシナリオをサポートします：

1. 名前にプレフィックスまたはサフィックスを追加する：定数を設定して、ソーステーブル名にプレフィックスまたはサフィックスを追加します。

   ルールの設定	結果
   [ルール名] フィールドに pre_table_post と入力します。[送信先インデックス名] は、テキスト pre_、[ソーステーブル名] 組み込み変数、およびテキスト _post の 3 つの部分で構成されます。	pre_table_post ルールを適用すると、ソーステーブル userinfo は送信先インデックス pre_userinfo_post (作成予定) にマッピングされ、ソーステーブル userinfo1 は送信先インデックス pre_userinfo1_post (作成予定) にマッピングされます。

2. グローバルな文字列置換を実行する：すべてのソーステーブル名に含まれる文字列 dev_ を prd_ に置き換えます。

   ルールの設定	結果
   [送信先インデックス名] ドロップダウンリストで、[ソーステーブル名] 組み込み変数を選択し、[組み込み変数の編集] ページに移動します。[ソーステーブル名] タブで、ソース文字列 dev_ を送信先文字列 prd_ に置き換える文字列置換ルールを設定します。ルールは上から下に実行されます。[上に移動]、[下に移動]、[削除] を使用して、ルールの順序を調整したり、削除したりできます。	結果は、ソーステーブル dev_table1 と dev_table2 が、[置き換え] ルールを使用して、送信先インデックス prd_table1 と prd_table2 (作成予定) にマッピングされることを示しています。dev_ プレフィックスは prd_ に置き換えられます。

3. 複数のテーブルから単一のテーブルにデータを書き込む：送信先インデックス名を定数値に設定します。

   ルールの設定	結果
   [送信先インデックス名] フィールドに、my_table のような固定値を入力して、すべてのソーステーブルを同じ送信先テーブルにマッピングします。	結果：ソースデータベース mysql_test2 のソーステーブル table_01 と table_02 は、両方とも同じ送信先インデックス my_table (作成予定) にマッピングされます。これにより、複数のテーブルのデータを単一のインデックスにマージできます。

フィールドデータ型のマッピングを編集 (任意)

システムは、[ソースタイプ] と [送信先タイプ] の間にデフォルトのマッピングを提供します。テーブルの右上隅にある Edit Mapping of Field Data Types をクリックして、ソーステーブルと送信先インデックス間のフィールドデータ型のマッピングをカスタマイズできます。設定が完了したら、Apply and Refresh Mapping をクリックします。

フィールドデータ型のマッピングを編集する際は、型変換ルールが有効であることを確認してください。そうしないと、型変換の失敗が発生し、ダーティデータが生成され、タスクが中断される可能性があります。

送信先インデックスを編集 (任意)

カスタムインデックス名マッピングルールに基づいて、システムは自動的に新しい送信先インデックスを作成するか、一致する名前の既存のインデックスを再利用します。

DataWorks は、ソーステーブルの構造に基づいて送信先インデックスの構造を自動的に生成します。ほとんどの場合、手動での介入は必要ありません。

送信先インデックスのステータスが [作成予定] の場合、元のテーブル構造に基づいて送信先インデックスに新しいフィールドを追加できます。次の操作を実行します：

1. 送信先インデックスにフィールドを追加します。

   • 単一のインデックスにフィールドを追加する: Destination Index Name 列の アイコンをクリックし、Statement Used to Create Index を編集してフィールドを追加します。

     • Dynamic Mapping Status：データ同期中にソーステーブルから新しいフィールドを送信先インデックスに追加するかどうかを指定します。有効な値：

       • true：システムがソーステーブルに新しいフィールドを検出した場合、そのフィールドを送信先インデックスに追加します。これらのフィールドは検索可能になります。これがデフォルト値です。

       • false：システムがソーステーブルに新しいフィールドを検出した場合、そのフィールドを送信先インデックスに追加しますが、これらのフィールドは検索できません。

       • strict：システムがソーステーブルに新しいフィールドを検出した場合、そのフィールドを送信先インデックスに追加することを拒否し、例外をスローします。エラーの詳細はログで確認できます。

       • runtime：システムがソーステーブルに新しいフィールドを検出した場合、新しいフィールドはインデックスマッピングに追加されません。代わりに、クエリ時にランタイムフィールドとして扱われます。これにより、フィールドをスクリプト計算や検索で使用できます。

       動的マッピングの詳細については、「動的マッピング」をご参照ください。

     • Shards と Replica Shards：インデックスのプライマリシャードとレプリカシャードの数。完全なインデックスは複数のシャードに分割され、異なる Elasticsearch ノードに分散されて、分散検索を可能にし、クエリパフォーマンスを向上させます。詳細については、「基本概念」をご参照ください。

       説明

       Shards と Replica Shards パラメーターの値は、タスク実行後に変更できません。両方のパラメーターのデフォルト値は 1 です。

   • フィールドの一括追加：同期するすべてのテーブルを選択し、テーブルの下部で Batch Modify > Destination Index Structure - Batch Add Fields を選択します。

Value assignment

ネイティブフィールドは、ソースと送信先の間で一致するフィールド名に基づいて自動的にマッピングされます。追加した 新しいフィールドと送信先インデックスのプロパティ には手動で値を割り当てる必要があります。次の操作を実行します：

• 単一のテーブルに値を割り当てる：Value assignment 列の Configuration ボタンをクリックして、送信先インデックスフィールドに値を割り当てます。

• 一括で値を割り当てる：リストの下部で Batch Modify > Value assignment を選択して、複数の送信先インデックスにまたがる同一のフィールドに同時に値を割り当てます。

Value Type を変更することで、定数と変数を割り当てることができます。以下のオプションがサポートされています：

• 送信先インデックスフィールド：

  • 手動割り当て：abc のような定数値を入力します。

  • ソースフィールド：ソーステーブルのフィールドから値を割り当てます。フィールド値または時間値のいずれかを選択できます。

    • フィールド値：ソースフィールドの値を直接送信先に書き込みます。

    • 時間値：ソースフィールドに時間値が含まれている場合、異なるフォーマットを使用して処理し、Destination Format を指定して抽出された値をフォーマットできます。

      • 時間文字列："2018-10-23 02:13:56" や "2021/05/18" のような時間や日付を表す文字列。時間フォーマットを指定することで、文字列は日付または時間値に解析されます。例えば、上記の例の文字列は yyyy-MM-dd HH:mm:ss と yyyy/MM/dd フォーマットで認識できます。

      • 時間オブジェクト：ソース値が Date や Datetime のような時間データ型の場合、この型を直接選択できます。

      • Unix タイムスタンプ (秒)：秒単位の 10 桁のタイムスタンプで、数値または文字列として提供できます。例：1610529203 および "1610529203"。

      • Unix タイムスタンプ (ミリ秒)：ミリ秒単位の 13 桁のタイムスタンプで、数値または文字列として提供できます。例：1610529203002 および "1610529203002"。

  • 変数を選択：システムが提供する変数を選択して値のソースとします。

  • 関数：関数を使用してソースフィールドに簡単な変換を適用してから、値として割り当てます。詳細については、「関数式を使用して送信先テーブルフィールドに値を割り当てる」をご参照ください。

• 送信先インデックスプロパティの割り当て：送信先インデックスのプライマリキーに値を割り当てます。複数のソースフィールドを連結して複合プライマリキーを作成できます。結果の値が一意であることを確認してください。

[ソース分割列]

[ソース分割列] ドロップダウンリストからソーステーブルのフィールドを選択するか、Not Split を選択できます。同期タスクが実行されると、この列に基づいて複数のサブタスクに分割され、データをバッチで並行して読み取ります。

テーブルのプライマリキーをソース分割列として使用することを推奨します。文字列、浮動小数点、日付型はサポートされていません。

ソース分割列は、ソースが MySQL の場合にのみサポートされます。

[フル同期をスキップ]

ステップ 3 でフル同期を設定した場合、個々のテーブルに対してフル同期をスキップすることを選択できます。これは、他の方法で既に完全なデータを送信先に同期している場合に便利です。

Full condition

フル同期フェーズ中にソースデータにフィルターを適用します。WHERE 句のフィルター条件のみを入力してください。WHERE キーワードは含めないでください。

Configure DML Rule

DML メッセージ処理は、ソースからキャプチャされた変更データ (Insert、Update、Delete) が送信先に書き込まれる前に、詳細なフィルタリングと制御を適用します。このルールは増分同期フェーズ中にのみ適用されます。