Data Integration を使用すると、単一の DataHub Topic、Hologres テーブル、Kafka Topic、または LogHub Logstore のデータを Hologres にリアルタイムで同期できます。単一の Hologres テーブルのデータを別の Hologres データソースに同期するためにリアルタイム同期タスクを実行すると、タスクはソース Hologres テーブルのスキーマに基づいて Hologres データソースにデスティネーションテーブルを作成し、ソーステーブルからデスティネーションテーブルにデータを同期します。このトピックでは、単一の Hologres テーブルのデータを別の Hologres テーブルに同期するためのリアルタイム ETL 同期タスクを作成する方法について説明します。
制限事項
Hologres インスタンスのバージョンは V2.1 以降である必要があります。
Hologres パーティションテーブルからのデータの増分同期はサポートされていません。
Hologres テーブルの DDL 変更メッセージは同期できません。
Hologres からは、次のデータ型の増分データを同期できます。
INTEGER、BIGINT、TEXT、CHAR(n)、VARCHAR(n)、REAL、JSON、SERIAL、OID、INT4[]、INT8[]、FLOAT8[]、BOOLEAN[]、および TEXT[]。
前提条件
サーバーレスリソースグループが購入済みであること。
2 つの Hologres データソースが DataWorks に追加されていること。1 つはソースとして使用され、もう 1 つはデスティネーションとして使用されます。詳細については、「Data Integration でのデータソースの追加と管理」をご参照ください。
サーバーレスリソースグループとデータソース間のネットワーク接続が確立されていること。詳細については、「ネットワーク接続ソリューション」をご参照ください。
単一の Hologres テーブルのデータを別の Hologres テーブルにリアルタイムで同期するには、ソース Hologres データベースの Hologres テーブルのバイナリロギング機能を有効にする必要があります。詳細については、「Hologres バイナリログの購読」をご参照ください。
手順
1. 同期タイプを選択する
Data Integration ページに移動します。
DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションウィンドウで、 を選択します。表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[Data Integration に移動] をクリックします。
Data Integration ページの左側のナビゲーションウィンドウで、[同期タスク] をクリックします。同期タスクページの上部で、ソースドロップダウンリストからソースタイプを、デスティネーションドロップダウンリストからデスティネーションタイプを選択し、[同期タスクの作成] をクリックします。表示されたページで、同期タスクの基本情報を構成します。
ソースとデスティネーション: ソースタイプとデスティネーションタイプの両方に
Hologresを選択します。新しいノード名: ビジネス要件に基づいて同期タスクの名前を指定します。
同期方法:
単一テーブルのリアルタイム同期を選択します。同期モード:
完全初期化を選択します。
2. ネットワーク設定とリソースグループを構成する
ネットワークとリソースの構成 セクションで、同期タスクの実行に使用する [リソースグループ] を選択し、[タスクリソースの使用量] パラメーターを構成します。
ソース セクションでソースとして、デスティネーション セクションでデスティネーションとして、追加された
Hologresデータソースをそれぞれ選択し、[接続テスト] をクリックします。
ネットワーク接続テストが成功したら、[次へ] をクリックします。
3. データ同期リンクを構成する
1. ソースを構成する
構成ページの上部にあるウィザードで、Hologres をクリックしてソースを構成します。
Holo ソース情報 セクションで、スキーマとテーブルのパラメーターを構成します。
[データサンプリング] セクションの右上隅にある [データサンプリング] をクリックします。
[データ出力のプレビュー] ダイアログボックスで、[サンプリングされたデータレコード] パラメーターを構成し、[収集開始] をクリックします。システムは、指定した Hologres テーブルからデータをサンプリングします。 Hologres テーブルのデータをプレビューできます。 Hologres テーブルのデータは、データ処理ノードのデータプレビューと可視化の構成への入力データとして使用されます。
2. データ処理ノードを構成する
アイコンをクリックして、データ処理方法を追加できます。サポートされているデータ処理方法は次のとおりです。 データマスキング、文字列の置換、データフィルタリング、JSON 解析、フィールドの編集と値の割り当て。ビジネス要件に基づいてデータ処理方法を調整できます。同期タスクが実行されると、指定した処理順序に基づいてデータが処理されます。
データ処理ノードを構成した後、セクションの右上隅にある [データ出力のプレビュー] をクリックできます。 データ出力のプレビューダイアログボックスで、[祖先ノードの出力を再取得] をクリックして、データ処理ノードが指定された Hologres テーブルからサンプリングされたデータを処理し、処理結果をプレビューできるようにします。
入力データがデータ処理ノードによって処理された後に生成された結果をプレビューする前に、Kafka データソースの [データサンプリング] 設定を構成する必要があります。
3. デスティネーションを構成する
構成ページの上部にあるウィザードで、Hologres をクリックしてデスティネーションを構成します。
デスティネーション情報 セクションで、スキーマとデスティネーションテーブルのパラメーターを構成します。 デスティネーションテーブルパラメーターの有効な値は、[テーブルを自動的に作成] と [既存のテーブルを使用] です。
デスティネーションテーブルパラメーターを テーブルを自動的に作成 に設定すると、システムはデスティネーションにソーステーブルと同じ名前のテーブルを自動的に作成します。作成されたテーブルの名前は手動で変更できます。
デスティネーションテーブルパラメーターを 既存のテーブルを使用 に設定すると、テーブル名ドロップダウンリストからテーブルを選択できます。
(オプション) デスティネーションテーブルのスキーマを変更します。
デスティネーションテーブルパラメーターで [テーブルを自動的に作成] を選択した場合は、[テーブルスキーマの編集] をクリックします。表示されたダイアログボックスで、自動的に作成されるデスティネーションテーブルのスキーマを編集します。 [祖先ノードの出力列に基づいてテーブルスキーマを再生成] をクリックして、祖先ノードの出力列に基づいてスキーマを再生成することもできます。生成されたスキーマから列を選択し、その列をプライマリキーとして構成できます。
説明デスティネーションテーブルにはプライマリキーが必要です。そうでない場合、構成を保存できません。
ジョブタイプ パラメーターと 書き込み競合ポリシー パラメーターを構成します。
ジョブタイプ パラメーターの有効な値:
再生 (操作ログを再生してデータを復元): ソースで実行されたのと同じ操作がデスティネーションで実行されることを示します。たとえば、ソースで INSERT、UPDATE、または DELETE 操作が実行された場合、デスティネーションでも同じ操作が実行されます。
挿入 (アーカイブストレージ): デスティネーションがストリーミングデータストレージとして使用され、ソースから同期されたすべてのデータがデスティネーションに挿入されることを示します。
書き込み競合ポリシー: データ書き込み競合が発生した場合に使用される処理ポリシー。有効な値: カバー (上書き) および 無視 (無視)。
ソースのフィールドとデスティネーションのフィールド間のマッピングを構成します。
上記の構成を完了すると、システムは [同一名マッピング] 原則に基づいて、ソースのフィールドとデスティネーションのフィールド間のマッピングを自動的に確立します。ビジネス要件に基づいてマッピングを変更できます。ソースの 1 つのフィールドをデスティネーションの複数のフィールドにマッピングできます。ソースの複数のフィールドをデスティネーションの同じフィールドにマッピングすることはできません。ソースのフィールドにデスティネーションにマッピングされたフィールドがない場合、ソースのフィールドのデータはデスティネーションに同期されません。
4. アラートルールを構成する
同期タスクの失敗によってビジネスデータの同期に遅延が発生するのを防ぐために、同期タスクに異なるアラートルールを構成できます。
ページの右上隅にある [アラートルールの構成] をクリックして、[アラートルールの構成] パネルに移動します。
アラートルールの構成パネルで、[アラートルールの追加] をクリックします。 アラートルールの追加ダイアログボックスで、パラメーターを構成してアラートルールを構成します。
説明この手順で構成するアラートルールは、同期タスクによって生成されるリアルタイム同期サブタスクに有効になります。同期タスクの構成が完了したら、「リアルタイム同期タスクの管理」を参照してリアルタイム同期タスクページに移動し、リアルタイム同期サブタスクに構成されたアラートルールを変更できます。
アラートルールを管理します。
作成されたアラートルールを有効または無効にできます。アラートの重大度レベルに基づいて、異なるアラート受信者を指定することもできます。
5. 詳細パラメーターを構成する
DataWorks では、特定のパラメーターの構成を変更できます。ビジネス要件に基づいて、これらのパラメーターの値を変更できます。
予期しないエラーやデータ品質の問題を防ぐために、パラメーターの値を変更する前に、パラメーターの意味を理解することをお勧めします。
構成ページの右上隅で、[詳細パラメーターの構成] をクリックします。
[高度なパラメーターの設定] パネルで、目的のパラメーターの値を変更します。
6. リソースグループを構成する
ページの右上隅にある [リソースグループの構成] をクリックして、現在の同期タスクの実行に使用されるリソースグループを表示および変更できます。
7. 同期タスクのテストを実行する
上記の構成が完了したら、構成ページの右上隅にある [シミュレート実行を実行] をクリックして、同期タスクがサンプリングされたデータをデスティネーションテーブルに同期できるようにします。デスティネーションテーブルで同期結果を表示できます。同期タスクの特定の構成が無効である場合、テスト実行中に例外が発生した場合、またはダーティデータが生成された場合、システムはリアルタイムでエラーを報告します。これは、同期タスクの構成を確認し、できるだけ早く予期される結果が得られるかどうかを判断するのに役立ちます。
表示されたダイアログボックスで、[開始日時] パラメーターと [サンプリングされたデータレコード] パラメーターなど、指定されたテーブルからのデータサンプリングのパラメーターを構成します。
[収集開始] をクリックして、同期タスクがソースからデータをサンプリングできるようにします。
[プレビュー] をクリックして、同期タスクがサンプリングされたデータをデスティネーションに同期できるようにします。
8. 同期タスクを実行する
同期タスクの構成が完了したら、ページの下部にある [完了] をクリックします。
[同期タスク] ページの [タスク] セクションで、作成された同期タスクを見つけ、[操作] 列の [開始] をクリックします。
[タスク] セクションで同期タスクの [名前または ID] をクリックし、同期タスクの詳細な実行プロセスを表示します。
同期タスクで O&M 操作を実行する
同期タスクの実行ステータスを表示する
同期タスクが作成された後、同期タスクページに移動して、ワークスペースに作成されたすべての同期タスクと各同期タスクの基本情報を表示できます。
操作列の [開始] または [停止] をクリックして、同期タスクを開始または停止できます。アクション列の [その他] をクリックし、[編集] または [表示] を選択して、同期タスクを変更したり、同期タスクに関する情報を表示したりすることもできます。
[実行の概要] 列に同期タスクの基本的な実行情報が表示されます。同期タスクの実行詳細ページのさまざまなセクションをクリックして、関連情報を表示することもできます。
同期タスクは 3 つのステージに分かれています。
[スキーママイグレーション]: このタブには、デスティネーションテーブルが自動的に作成されたテーブルか既存のテーブルかなどの情報が表示されます。自動的に作成されたテーブルの場合、テーブルの作成に使用された DDL 文が表示されます。
[完全データ初期化]: 同期タスクの構成時に [同期モード] パラメーターを [完全初期化] に設定した場合、完全同期の進行状況がこのセクションに表示されます。
[リアルタイムデータ同期]: このタブには、リアルタイムの読み取り/書き込みトラフィック、ダーティデータ情報、フェールオーバー、操作ログなど、リアルタイム同期の統計が表示されます。
同期タスクを再実行する
特別な場合に、同期するフィールド、デスティネーションテーブルのフィールド、またはテーブル名情報を変更する場合、目的の同期タスクの [操作] 列にある [再実行] をクリックすることもできます。このようにして、システムはデスティネーションに加えられた変更を同期します。すでに同期されていて変更されていないテーブルのデータは再度同期されません。
同期タスクの構成を変更せずに [再実行] を直接クリックすると、システムは同期タスクを再実行できます。
同期タスクの構成を変更してから、[完了] をクリックします。同期タスクの操作列に表示される [更新の適用] をクリックして、最新の構成を有効にするために同期タスクを再実行します。