データソース、ネットワーク、リソースを設定した後、リアルタイム同期タスクを作成して、単一テーブルまたはデータベース全体のデータを同期できます。このトピックでは、単一テーブルまたはデータベース全体の増分データに対してリアルタイム同期タスクを作成し、作成後にタスクのステータスを表示する方法について説明します。
前提条件
データソースの設定が完了していること。タスクを作成する前に、ソースデータベースとターゲットデータベースを設定する必要があります。これにより、タスク設定時に読み取りと書き込みに必要なデータソースを選択できるようになります。リアルタイム同期でサポートされているデータソースと設定の詳細については、「サポートされているデータソースと同期ソリューション」をご参照ください。
適切な仕様のリソースグループを購入し、ワークスペースにアタッチしていること。詳細については、「Serverless リソースグループの使用」をご参照ください。
リソースグループとデータソース間のネットワーク接続が利用可能であること。詳細については、「ネットワーク接続の設定」をご参照ください。
制限事項
このバージョンのリアルタイム同期は、DataStudio (旧バージョン) でのみサポートされています。
DataStudio への移動
一部のチャンネルでは、DataStudio で単一テーブルのリアルタイム同期タスクを作成する必要があります。チャンネルのサポート状況の詳細については、「サポートされているデータソース」をご参照ください。
DataWorks コンソールにログインします。目的のリージョンを選択し、左側のナビゲーションウィンドウで をクリックします。ドロップダウンリストから目的のワークスペースを選択し、[DataStudio へ] をクリックします。
ステップ 1：タスクの作成
DataStudio (新バージョン)
ワークフローを作成します。詳細については、「ワークフロー」をご参照ください。
リアルタイム同期ノードを作成します。次のいずれかの方法を使用できます。
方法 1：ワークフローリストの右上隅にある
アイコンをクリックし、 を選択します。方法 2: ワークフロー名をダブルクリックし、左側のノードリストで [データ統合] ディレクトリを展開してから、[リアルタイム同期] ノードを右側のキャンバスにドラッグします。
「[Create Node]」ダイアログボックスで、ノードのパラメーターを設定し、「[OK]」をクリックします。
DataStudio (旧バージョン)
ビジネスフローを作成します。詳細については、「ワークフローの作成」をご参照ください。
リアルタイム同期タスクを作成します。次のいずれかの方法を使用できます。
方法 1: ビジネスフローを展開し、を右クリックします。
方法 2: ビジネスフロー名をダブルクリックし、[ノードの作成] をクリックしてから、[リアルタイム同期] ノードを [データ統合] ディレクトリから右側のビジネスフロー編集パネルにドラッグします。
[ノードの作成] ダイアログボックスで、パラメーターを設定します。
パラメーター | 説明 |
タイプ | デフォルトは [リアルタイム同期] です。 |
[同期方式] |
|
パス | リアルタイム同期タスクが保存されるディレクトリです。 |
名前 | ノード名には、英字、漢字、数字、アンダースコア (_)、ピリオド (.) を使用できます。128 文字を超えることはできません。 |
ステップ 2：リソースグループの設定
リアルタイム同期タスクは、Serverless リソースグループまたは Data Integration 専用リソースグループのみを使用できます。リアルタイム同期タスクの編集ページの右側のナビゲーションバーにある[基本設定]をクリックし、[リソースグループ] ドロップダウンリストで、データベースへのネットワーク接続があるリソースグループを選択します。
リソースグループを作成したにもかかわらず表示されない場合は、リソースグループがワークスペースにアタッチされているかどうかを確認してください。詳細については、「Serverless リソースグループの使用」をご参照ください。
リソースの競合を防ぐために、リアルタイム同期タスクとオフライン同期タスクは異なるリソースグループで実行することを推奨します。CPU、メモリ、ネットワークリソースの競合は、オフラインタスクの速度低下、リアルタイムタスクの遅延、または極端な場合には OOM (Out of Memory) キラーによるタスクの終了を引き起こす可能性があります。
Serverless リソースグループは、同期タスクの CU 上限の指定をサポートしています。リソース不足により同期タスクで OOM の問題が発生した場合は、リソースグループの CU 使用量を適切に調整してください。
ステップ 3：タスクの設定
単一テーブル同期の設定
入力データソースを設定します。
現在、単一テーブルデータ同期でサポートされている入力データソースのタイプと設定は次のとおりです。
左側の [入力] リストから入力データソースコンポーネントを右側のキャンバスにドラッグします。
入力コンポーネントをクリックし、右側の [ノード設定] ダイアログボックスで関連情報を設定します。
(オプション)データ変換を設定します。
データ変換を設定して、入力データを変更できます。
現在、単一テーブルデータ同期でサポートされている変換方法は次のとおりです。
データフィルタリング変換の設定：ルールに基づいてデータをフィルタリングします。一致するデータのみが保持されます。
文字列置換の設定：文字列型のフィールドの置換を許可します。
データマスキングの設定：単一テーブルのリアルタイム同期中に機密データをマスキングし、指定されたデータベースの場所に保存できます。
リアルタイム同期タスク編集ページの左側にある[変換]エリアで、必要なデータ変換コンポーネントを右側のパネルにドラッグします。入力コンポーネントの上にカーソルを合わせて、その接続ポイントを表示します。入力コンポーネントの下部接続ポイントを変換コンポーネントの上部接続ポイントに接続します。接続後、変換コンポーネントに対して[ノード設定]を実行できます。
変換コンポーネントをクリックし、右側の[ノード設定]ダイアログボックスで関連情報を設定します。
出力データソースを設定します。
現在、単一テーブルデータ同期でサポートされている出力データソースのタイプと設定は次のとおりです。
リアルタイム同期タスク編集ページの左側にある[出力]エリアで、送信先出力データソースコンポーネントを右側のパネルにドラッグし、上流コンポーネントに接続します。送信先データソース、テーブル、およびフィールドマッピング関係を設定します。送信先テーブルが存在しない場合は、[テーブルの作成]をクリックして、すぐにテーブルを作成します。
出力コンポーネントをクリックし、右側の[ノード設定]ダイアログボックスで関連情報を設定します。
キャンバスの上部にあるツールバーで、[保存] をクリックして、タスクの構成を完了します。
データベース全体の同期の設定
DataWorks は、データ統合のリアルタイムデータベース同期タスクの使用を推奨します。
同期ソースとルールを設定します。
[データソース] エリアで、同期する [タイプ] および [データソース] の名前を選択します。
同期するテーブルを選択します。
[同期するソーステーブルの選択] セクションで、[ソーステーブル] リストから同期するテーブルを選択し、
アイコンをクリックして [選択したテーブル] リストに移動します。重要選択したテーブルにプライマリキーがない場合、リアルタイム同期は実行できません。
テーブル名のマッピングルールを設定します。
このステップでは、ソース データソースから同期するデータベースおよびテーブルを選択できます。デフォルトでは、同期ソリューションは、ソース データベースおよびテーブルのデータを、ターゲットで同じ名前のスキーマまたはテーブルに書き込みます。ターゲットに該当するスキーマまたはテーブルが存在しない場合、自動的に作成されます。また、[テーブル名のマッピングルール] を使用して、ターゲットに書き込まれる最終的なスキーマまたはテーブル名を定義することもできます。これにより、複数のテーブルからのデータを単一のテーブルに書き込んだり、ターゲット データベースまたはテーブルに書き込む際に、ソース データベースまたはテーブル名の固定プレフィックスを一括で更新したりできます。
[ソーステーブル名とターゲットテーブル名の変換ルール]：正規表現を使用して、ソーステーブル名を最終的なターゲットテーブル名に変換することをサポートします。
例 1：ソースで「doc_」プレフィックスを持つテーブルのデータを、ターゲットで「pre_」プレフィックスを持つテーブルに書き込みます。
例 2：複数のテーブルのデータを単一のターゲットテーブルに書き込みます。
ソースのテーブル "table_01"、"table_02"、"table_03" を、宛先テーブル "my_table" に同期します。正規表現のテーブル名変換ルールを次のように設定します: ソース: table.*、宛先: my_table。
[ターゲットテーブル名のルール]：組み込み変数の組み合わせを使用してターゲットテーブル名を生成することをサポートします。変換されたターゲットテーブル名にプレフィックスとサフィックスを追加することもできます。利用可能な組み込み変数は次のとおりです。
${db_table_name_src_transed}: ソーステーブル名および宛先テーブル名の変換ルール での変換後のテーブル名。
${db_name_src_transed}：「ソースデータベースとターゲットスキーマ名の変換ルール」で変換された後のターゲットスキーマ名。
${ds_name_src}：ソースデータソース名。
例：前のステップで変換されたテーブル名を文字列連結でさらに処理します。前のステップの結果「my_table」を表すために ${db_table_name_src_transed} を使用し、この組み込み変数にプレフィックスとサフィックスを追加します。たとえば、pre_${db_table_name_src_transed}_post のようにします。これにより、ターゲットテーブルは「pre_my_table_post」という名前になります。
[ソースデータベース名とターゲットスキーマ名の変換ルール]：正規表現を使用して、ソーススキーマ名を最終的なターゲットスキーマ名に変換することをサポートします。
例：ソースデータベース名の「doc_」プレフィックスを「pre_」プレフィックスに置き換えます。
ターゲットデータソースを選択し、ターゲットテーブルまたは Topic を設定します。
「[送信先テーブルの設定]」ページで、送信先の基本情報を設定します。たとえば、書き込みモードやパーティション設定などです。具体的な設定内容は、各データソースのリアルタイム同期インターフェイスによって異なります。
[ソーステーブルと宛先テーブルのマッピングの更新] をクリックして、同期するソーステーブルと宛先テーブル間のマッピングを作成します。
[追加フィールドの編集] で、宛先スキーマと宛先テーブル名をカスタマイズし、宛先テーブルに定数または変数を追加できます。具体的な構成は、各データソースのリアルタイム同期インターフェイスによって異なります。
説明同期するテーブルが多い場合、プロセスが遅くなることがあります。しばらくお待ちください。
(オプション)テーブル粒度の同期ルールを設定します。
一部の同期ソリューションは、カスタムのテーブルレベル DML 処理ポリシーをサポートしています。ソーステーブルでの insert、update、delete 操作の処理ポリシーを定義できます。
説明サポートされる DML 操作はデータソースによって異なる場合があります。特定の同期ソリューションが DML 処理ポリシーをサポートしているかどうかは、プロダクトのインターフェイスをご参照ください。データソース別の現在の DML サポートについては、「サポートされている DML および DDL 操作」をご参照ください。
DDL メッセージ処理ルールを設定します。
同期中の DDL 操作を処理するポリシーを設定できます。サポートされる DDL 操作はデータソースによって異なる場合があります。詳細については、「サポートされている DML および DDL 操作」をご参照ください。 ページで、各ターゲットデータベースタイプの DDL 処理ポリシーを設定できます。さまざまな DDL メッセージ処理ポリシーを次の表に示します。
DDL メッセージタイプ
処理ポリシー
[テーブルの作成]
DataWorks が対応するタイプの DDL メッセージを受信した場合の処理ポリシーは次のとおりです。
[通常処理]：メッセージをターゲットデータソースに転送します。
[無視]：メッセージを直接破棄し、ターゲットデータソースに送信しません。
[アラート]：メッセージを直接破棄し、実行エラーによりメッセージが破棄されたことを示すアラートをリアルタイム同期ログに記録します。
[エラー]：リアルタイム同期タスクは直接エラーステータスを表示し、実行を終了します。
テーブル削除
[列の追加]
カラムの削除
[テーブル名の変更]
[列名の変更]
[列タイプの変更]
テーブルの切り捨て
実行リソースを設定します。
データ統合の最大同時読み取りおよび書き込み数を制限するための同時実行制御を提供します。
同期タスクがダーティデータを許容するかどうかを制御することをサポートします。
ダーティデータが許可されない場合：タスク実行中にダーティデータが生成されると、タスクは失敗して終了します。
ダーティデータが許可される場合：同期タスクはダーティデータを無視し (ターゲットには書き込まれません)、正常に実行を継続します。
［設定の完了］をクリックします。
ステップ 4：ノードの送信とデプロイ
ツールバーの
アイコンをクリックしてノードを送信します。「[Submit]」ダイアログボックスで、「[Change Description]」を入力します。
[OK] をクリックします。
標準モードのワークスペースを使用している場合、タスクが正常に提出された後、本番環境にデプロイする必要があります。上部のメニューバーの左側にある[デプロイ]をクリックします。詳細については、「タスクを公開する」をご参照ください。
ステップ 5：タスクの実行
リアルタイム同期タスクは DataStudio では実行できません。タスクをオペレーションセンターに公開して開始し、表示してください。
タスクの設定が完了したら、 パネルでタスクを開始および管理できます。詳細については、「リアルタイム同期タスクの O&M」をご参照ください。
次のステップ
タスクが開始されたら、タスク名をクリックして実行の詳細を表示し、タスクの O&M とチューニングを実行できます。
よくある質問
リアルタイム同期タスクに関する一般的な質問については、「リアルタイム同期」をご参照ください。
付録：タスクの移行
DataStudio から単一テーブルのリアルタイム統合タスクを [Data Integration] ページに移行するには、[Data Integration への移行] をクリックします。
現在サポートされているリアルタイム統合タスク：
Kafka から MaxCompute への単一テーブルリアルタイム統合タスク。
Kafka から Hologres への単一テーブルリアルタイム統合タスク。
移行対象の単一テーブル リアルタイム統合タスクをダブルクリックしてタスク編集ページを開き、[データ統合への移行] をクリックしてタスクを移行します。
左上隅の
をクリックし、 を選択します。[同期タスク] ページに移動して、タスク一覧で正常に移行された単一テーブルのリアルタイム統合タスクを表示します。
移行されたタスクは、オペレーションセンターに移動することなく、データ統合のメインサイトで直接メンテナンスできます。移行は、保存されたタスク設定や実行中のタスクには影響しません。
移行後、元のタスクは DataStudio のごみ箱に移動されます。その後の編集およびメンテナンス操作は、データ統合のメインサイトのタスクリストページでのみ実行できます。