DataWorks:データの同期

DataStudio

DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションウィンドウで、[データ開発と O&M] > [データ開発] を選択します。表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[データ開発へ進む] をクリックします。

ワークフローの設計

1. ワークフローを作成します。

   DataWorks では、ワークフロー内のコンポーネントを使用してデータ開発を行います。ノードを作成する前に、ワークフローを作成する必要があります。手順については、「ワークフローの作成」をご参照ください。

   ワークフローに user_profile_analysis_MaxCompute という名前を付けます。

   

2. ワークフローを設計します。

   ワークフローを作成すると、そのキャンバスが自動的に開きます。キャンバスで New Node をクリックします。ノードをキャンバスにドラッグし、それらを接続して依存関係を定義することで、データ同期ワークフローを設計します。詳細については、「ワークフロー設計」をご参照ください。

   

3. このチュートリアルでは、ゼロロードノードとバッチ同期ノードの間にデータリネージは存在しません。したがって、ワークフローキャンバスに線を描画して依存関係を定義します。依存関係の設定方法の詳細については、「スケジューリング依存関係の設定ガイド」をご参照ください。以下の表に、各ノードのタイプ、名前、および機能を示します。

   カテゴリ	タイプ	名前 (最終出力テーブルにちなんで命名)	説明
   一般	ゼロロードノード	workshop_start_odps	ユーザープロファイル分析ワークフローを編成します。たとえば、開始時刻を設定します。複雑なワークフローでのデータフローパスを明確にします。このノードはドライランタスクであり、コード編集は不要です。
   データ統合	バッチ同期	ods_user_info_d_odps	MySQL から MaxCompute テーブル ods_user_info_d_odps に基本ユーザー情報を同期します。
   データ統合	バッチ同期	ods_raw_log_d_odps	OSS から MaxCompute テーブル ods_raw_log_d_odps にユーザーの Web サイトアクセスログを同期します。

スケジューリングロジックの設定

このチュートリアルでは、workshop_start_odps ゼロロードノードが毎日 00:30 にワークフローを実行するようにトリガーします。以下の表に、このノードの主要なスケジューリング構成の詳細を示します。他のノードのスケジューリング設定を変更する必要はありません。実装の詳細については、「時間プロパティの詳細設定」をご参照ください。他のスケジューリング構成の詳細については、「タスクスケジューリングプロパティ構成の概要」をご参照ください。

MaxCompute テーブルの作成

開始する前に、Data Integration が同期する生データを格納するターゲットの MaxCompute テーブルを作成します。このチュートリアルでは、それらを迅速に作成する方法を説明します。詳細な手順については、「MaxCompute テーブルの管理」をご参照ください。

1. テーブル作成ページに移動します。

   

2. ods_raw_log_d_odps テーブルを作成します。

   [テーブルの作成] ダイアログボックスで、[名前] フィールドに ods_raw_log_d_odps と入力します。テーブルエディターページで、[DDL] をクリックし、次の文を入力してから Generate Table Schema をクリックします。Confirm をクリックして、現在のスキーマを上書きします。

   CREATE TABLE IF NOT EXISTS ods_raw_log_d_odps
   (
    col STRING
   ) 
   PARTITIONED BY
   (
    dt STRING
   )
   LIFECYCLE 7;

3. ods_user_info_d_odps テーブルを作成します。

   [テーブルの作成] ダイアログボックスで、[名前] フィールドに ods_user_info_d_odps と入力します。テーブルエディターページで、[DDL] をクリックし、次の文を入力してから Generate Table Schema をクリックします。Confirm をクリックして、現在のスキーマを上書きします。

   CREATE TABLE IF NOT EXISTS ods_user_info_d_odps (
    uid STRING COMMENT 'ユーザー ID',
    gender STRING COMMENT '性別',
    age_range STRING COMMENT '年齢層',
    zodiac STRING COMMENT '星座'
   )
   PARTITIONED BY (
    dt STRING
   )
   LIFECYCLE 7;

4. テーブルをコミットして公開します。

   テーブル情報が正しいことを確認した後、ods_user_info_d_odps と ods_raw_log_d_odps の各テーブルについて、Commit to Development Environment をクリックし、次に Commit to Production Environment をクリックします。この操作により、ノード構成で定義された開発環境と本番環境のコンピュートエンジンプロジェクトに物理テーブルが作成されます。

   説明

   テーブルスキーマを定義した後、テーブルを開発環境と本番環境の両方にコミットする必要があります。テーブルが正常にコミットされると、対応するコンピュートエンジンプロジェクトで表示できます。

   • テーブルを DataWorks 開発環境にコミットして、開発環境に関連付けられた MaxCompute エンジンにテーブルを作成します。

   • テーブルを DataWorks 本番環境にコミットして、本番環境に関連付けられた MaxCompute エンジンにテーブルを作成します。

データソースの作成

このチュートリアルでは、ApsaraDB for RDS (MySQL) インスタンスと OSS のテストデータを使用します。提供されたテストデータにアクセスするには、ワークスペースに user_behavior_analysis_mysql という名前の MySQL データソースと user_behavior_analysis_httpfile という名前の HttpFile データソースを作成する必要があります。データソースに必要な基本情報が提供されています。

ユーザーデータ同期の設定

このバッチ同期ノードは、テストケースで提供される MySQL テーブル ods_user_info_d から MaxCompute テーブル ods_user_info_d_odps に基本ユーザー情報データを同期します。

1. バッチ同期ノード ods_user_info_d_odps をダブルクリックして、ノード編集ページを開きます。

2. ネットワークとリソースの設定。

   Data source、My Resource Group、および Data going を設定し、次のステップ をクリックします。プロンプトに従って接続テストを実行します。以下の表にこれらの設定を示します。

   

   パラメーター	説明
   Data source	Data source：MySQL Data Source Name：user_behavior_analysis_mysql
   My Resource Group	「環境の準備」ステップで購入したサーバーレスリソースグループを選択します。
   Data going	Data going：MaxCompute Data Source Name：user_behavior_analysis_odps

3. タスクの設定。

   • ソースと送信先を設定します。

     モジュール	パラメーター	説明	図
     Data source	Table	MySQL テーブル ods_user_info_d を選択します。
     	Shard Key	プライマリキーまたはインデックス付きの列を分割キーとして使用します。分割キーは整数フィールドである必要があります。 ここでは、分割キーを uid フィールドに設定します。
     Data going	Tunnel Resource Group	このチュートリアルでは、デフォルトの [共通転送リソース] を使用します。専用のトンネルクォータがある場合は、ドロップダウンリストから選択できます。 説明 MaxCompute データ転送リソースの詳細については、「専用データ転送リソースグループの購入と使用」をご参照ください。支払い遅延や有効期限切れにより専用のトンネルクォータが利用できない場合、タスクはランタイム中に自動的に [共通転送リソース] に切り替わります。
     	スキーマ	このチュートリアルでは、default を選択します。MaxCompute プロジェクトに他のスキーマがある場合は、ドロップダウンリストから選択できます。
     	Table	アドホッククエリで作成された ods_user_info_d_odps テーブルを選択します。
     	Partition Information	このチュートリアルでは、${bizdate} パラメーターを使用します。
     	Write Method	このチュートリアルでは、ドロップダウンリストから「書き込み前に既存のデータをクリーンアップ」を選択します。 2つの書き込みモードが利用可能です： insert into：テーブルまたは静的パーティションにデータを直接挿入します。 insert overwrite：新しいデータを挿入する前に、テーブルまたは静的パーティションから既存のデータをクリアします。
     	Write by Converting Empty Strings into Null	このチュートリアルでは、No を選択します。

   • フィールドマッピングと一般設定の確認。

     DataWorks では、ソースとターゲット間でフィールドをマッピングできます。また、読み取り/書き込みの同時実行数、データベースへの影響を避けるためのレート制限、ダーティデータの処理、分散タスクの実行などの設定も構成できます。このチュートリアルでは、デフォルト設定を使用します。他の構成オプションの詳細については、「ウィザードモードの設定」をご参照ください。

4. スケジューリングプロパティ。

   ノード設定ページで、右側のナビゲーションウィンドウの [プロパティ] をクリックして Scheduling Configuration パネルを開きます。このパネルで、スケジューリングとノード情報を設定します。詳細については、「ノードスケジューリング設定」をご参照ください。以下の表に構成を示します。

   パラメーター	説明	図
   Scheduling Parameters	スケジューリングパラメーターの値は、デフォルトの $bizdate のままにしておくことができます。 説明 bizdate=$bizdate は、前日の日付を yyyymmdd 形式で返します。
   Time attribute	Scheduling period：日。 Scheduling Time：00:30。 RUN Attribute：実行ステータスに関係なく許可。 他のパラメーターはデフォルト値のままにします。 説明 このノードの毎日の開始時刻は、workshop_start 仮想ノードのスケジュール時刻に依存します。ノードは毎日 00:30 以降にのみ実行されます。
   Resource Group	「環境の準備」ステップで作成したサーバーレスリソースグループを選択します。
   Scheduling Dependency	Parent Nodes を確認します：workshop_start ノードが現在のノードの上流ノードとして表示されていることを確認します。線を描画して設定した上流の依存関係がここに表示されます。workshop_start ノードが表示されない場合は、「ビジネスフローの設計」で説明されているように、データ同期段階のビジネスフロー設計を完了していることを確認してください。 この例では、workshop_start ノードのスケジュール時刻に達し、その実行が完了すると、現在のノードが実行されるようにトリガーされます。 [このノードの出力] として、「MaxCompute_production_project_name.ods_user_info_d_odps」という名前のノード出力が存在するかどうかを確認します。このノード出力が存在しない場合は、Node Output を手動で追加します。 説明 DataWorks は、ノード出力を使用してノードの依存関係を確立します。下流の SQL タスクが同期タスクの出力テーブルを処理しやすくするために、自動解析メカニズムを使用して、テーブルリネージに基づいて依存関係を迅速に追加できます。この場合、同期出力テーブルと同じ名前の ノード出力が存在することを確認する必要があります。たとえば、MaxCompute_production_project_name.ods_user_info_d_odps のようになります。

ユーザーログパイプラインの設定

このバッチ同期ノードは、パブリックな HttpFile オブジェクト user_log.txt から MaxCompute テーブル ods_raw_log_d_odps にユーザーの Web サイトアクセス情報を同期するタスクを設定するために使用されます。

1. バッチ同期ノード ods_raw_log_d_odps をダブルクリックして、ノード編集ページを開きます。

2. ネットワークとリソースの設定。

   Data source、My Resource Group、および Data going を設定します。次に、次のステップ をクリックし、プロンプトに従って接続テストを実行します。以下の表に構成を示します。

   

   パラメーター	説明
   Data source	Data source： HttpFile Data Source Name：user_behavior_analysis_HttpFile
   My Resource Group	「環境の準備」ステップで購入したサーバーレスリソースグループを選択します。
   Data going	Data going：MaxCompute Data Source Name：user_behavior_analysis_odps

3. タスクの設定。

   • ソースと送信先を設定します。

     モジュール	パラメーター		説明	図
     Data source	File Path		このチュートリアルでは、/user_log.txt と入力します。
     	File Type		text タイプを選択します。
     	Field Delimiter		このチュートリアルでは、| と入力します。
     	Advanced Configuration	Encoding	UTF-8 エンコード形式を選択します。
     		Compression Format	None 形式を選択します。
     		Skip Header	テーブルヘッダーをスキップしないように No を選択します。
     Data going	Tunnel Resource Group		このチュートリアルでは、デフォルトで [共通転送リソース] を使用します。専用のトンネルクォータがある場合は、ドロップダウンリストから選択できます。 説明 MaxCompute のデータ転送リソースの選択については、「専用データ転送サービスリソースグループの購入と使用」をご参照ください。支払い遅延や有効期限切れにより専用のトンネルクォータが利用できない場合、タスクは実行中に自動的に [共通転送リソース] に切り替わります。
     	スキーマ		このチュートリアルでは、デフォルトのスキーマを選択します。DataWorks プロジェクトに他のスキーマがある場合は、ドロップダウンリストから選択できます。
     	Table		ドロップダウンリストから、アドホッククエリで作成された ods_raw_log_d_odps テーブルを選択します。
     	Partition Information		このチュートリアルでは、${bizdate} パラメーターを使用します。
     	Write Method		このチュートリアルでは、[書き込み前に既存のデータをクリーンアップ] を選択します。 以下の書き込みモードがサポートされています： insert into：テーブルまたは静的パーティションにデータを直接挿入します。 insert overwrite：新しいデータを挿入する前に、テーブルまたは静的パーティションから既存のデータをクリアします。
     	Write by Converting Empty Strings into Null		このチュートリアルでは、No を選択します。

     ソースを設定した後、Confirm Data Structure をクリックして、DataWorks がログファイルを正しく読み取れることを確認します。

   • フィールドマッピングと一般設定の確認。

     DataWorks では、ソースフィールドをターゲットフィールドにマッピングできます。また、並列処理、レート制限 (データベースへの影響を避けるため)、ダーティデータの処理、分散タスクの実行などの機能も提供しています。このチュートリアルでは、デフォルト設定を使用します。他の構成オプションの詳細については、「ウィザードモードでバッチ同期タスクを設定する」をご参照ください。

4. スケジューリングプロパティの設定。

   設定ページで、右側のパネルの [スケジューリング設定] をクリックします。次に、Scheduling Configuration パネルで、スケジューリングとノード情報を設定します。詳細については、「ノードのスケジューリングプロパティを設定する」をご参照ください。以下の表に構成を示します。

   パラメーター	説明	図
   Scheduling Parameters	スケジューリングパラメーターの値は、デフォルトの $bizdate のままにしておくことができます。 説明 bizdate=$bizdate は、前日の日付を yyyymmdd 形式で返します。
   Time attribute	Scheduling period：日。 Scheduling Time：00:30。 RUN Attribute：実行ステータスに関係なく許可。 他のパラメーターはデフォルト値のままにします。 説明 このノードの毎日の開始時刻は、workshop_start 仮想ノードのスケジュール時刻に依存します。ノードは毎日 00:30 以降にのみ実行されます。
   Resource Group	「環境の準備」段階で作成したサーバーレスリソースグループを選択します。
   Scheduling Dependency	Parent Nodes を確認します：現在のノードの上流ノードとして workshop_start ノードが表示されていることを確認します。ノードを接続して設定された上流の依存関係がこのセクションに表示されます。workshop_start ノードが表示されない場合は、「ビジネスフローの設計」で説明されているように、ビジネスデータ同期段階のビジネスフロー設計を完了していることを確認してください。 この例では、workshop_start ノードのスケジュール時刻に達し、その実行が完了すると、現在のノードが実行されるようにトリガーされます。 [このノードの出力] を確認します：YourMaxComputeProjectName.ods_raw_log_d_odps という名前のノード出力が存在するかどうかを確認します。ノード出力が存在しない場合は、手動で Node Output を追加する必要があります。 説明 DataWorks は、ノード出力を使用してノードの依存関係を確立します。下流の SQL タスクが同期タスクの出力テーブルを処理する際、自動解析メカニズムは、テーブルリネージを使用して SQL ノードの依存関係として同期タスクを迅速に追加します。このメカニズムが機能するためには、同期出力テーブルと同じ名前の ノード出力が存在することを確認する必要があります：MaxCompute 本番プロジェクト名.ods_raw_log_d_odps。

ビジネスフローの実行

1. [DataStudio] ページで、ビジネスプロセス の下にある user_profile_analysis_MaxCompute をダブルクリックしてワークフローパネルを開き、ツールバーの アイコンをクリックして、上流と下流の依存関係に従ってワークフローを実行します。

2. 実行ステータスを確認します：

   • ノードステータスの確認： ステータスは、同期実行プロセスに問題がないことを示します。

   • タスク実行ログを表示するには、ods_user_info_d_odps または ods_raw_log_d_odps ノードを右クリックし、[ログの表示] を選択します。ログに次のテキストが表示された場合、同期ノードが正常に実行され、データが正常に同期されたことを示します。

     

同期結果の表示

ビジネスワークフローを実行すると、MySQL の基本ユーザー情報テーブル ods_user_info_d のデータが workshop2024_01_dev.ods_user_info_d_odps テーブルの昨日のパーティションに完全に同期され、OSS の user_log.txt ファイルのユーザー Web サイトアクセスログが workshop2024_01_dev.ods_raw_log_d_odps テーブルの昨日のパーティションに同期されることが期待されます。クエリ SQL は本番環境で実行する必要がないため、アドホッククエリファイルを作成して結果を確認できます。

1. アドホッククエリファイルを作成します。

   DataStudio ページの左側のナビゲーションウィンドウで、 アイコンをクリックして [アドホッククエリ] ペインを開きます。[アドホッククエリ] を右クリックし、New Node > [ODPS SQL] を選択します。

2. 同期結果テーブルをクエリします。

   次の SQL ステートメントを実行して、データ同期の結果を確認します。ods_raw_log_d_odps と ods_user_info_d_odps にインポートされたレコード数を確認します。

   // 'your_data_timestamp' を実際のデータタイムスタンプに置き換えます。これはタスク実行日の前日です (例：タスクが 20230621 に実行される場合は 20230620 を使用します)。
   select count(*) from ods_user_info_d_odps  where dt='your_data_timestamp'; 
   select count(*) from ods_raw_log_d_odps where dt='your_data_timestamp';

   

   説明

   このチュートリアルでは、開発環境である [DataStudio] でノードを実行します。したがって、データはデフォルトで、この環境のコンピュートエンジンプロジェクトである workshop2024_01_dev プロジェクトのテーブルに書き込まれます。