DataWorks:データ同期

DataStudio への移動

DataWorks コンソールにログインします。 左側のナビゲーションウィンドウで、[データモデリングと開発] > [DataStudio] を選択します。 表示されるページで、ドロップダウンリストから目的のワークスペースを選択し、[DataStudioに移動] をクリックします。

ワークフローの設計

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

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

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

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

   ワークフローを作成すると、そのキャンバスが自動的に開きます。ワークフロー設計に基づき、キャンバスで ノードの作成 をクリックします。必要なノードをキャンバスにドラッグし、ノード間に線を引いて、データ同期ステージの依存関係を設定します。

   

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 テーブルを事前に作成する必要があります。このチュートリアルでは、必要なテーブルを迅速に作成する方法を示します。詳細については、「Create and use MaxCompute tables」をご参照ください。

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

   ナビゲーションペインで、MaxCompute の [Table] ノードを右クリックし、ショートカットメニューから [Create Table] を選択します。

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

   [Create Table] ダイアログボックスで、名前に ods_raw_log_d_odps を入力します。テーブルエディターページで DDL パターン をクリックし、以下の DDL 文を入力します。次に テーブルスキーマの生成 をクリックし、確認 をクリックして現在の構成を上書きします。

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

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

   [Create Table] ダイアログボックスで、名前に ods_user_info_d_odps を入力します。テーブルエディターページで DDL パターン をクリックし、以下の DDL 文を入力します。次に テーブルスキーマの生成 をクリックし、確認 をクリックして現在の構成を上書きします。

   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 の両方のテーブルに対して 開発環境へのコミット と 本番環境へのコミット をクリックします。この操作により、ノード構成に基づいて、開発環境と本番環境に対応するコンピューティングエンジンプロジェクトに物理テーブルが作成されます。

   説明

   コミットが成功した後にのみ、対応する環境のコンピューティングエンジンプロジェクトでテーブルを確認できます。

   • テーブルを DataWorks の開発環境にコミットすると、開発環境に関連付けられている MaxCompute コンピューティングエンジンにテーブルが作成されます。

   • テーブルを DataWorks の本番環境にコミットすると、本番環境に関連付けられている MaxCompute コンピューティングエンジンにテーブルが作成されます。

ソースデータソースの作成

このチュートリアルでは、MySQL database と Object Storage Service (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. ネットワークとリソースを構成します。

   データソース、マイリソースグループ、データ宛先 を構成した後、次へ をクリックし、画面の指示に従って接続テストを完了します。必要な構成を以下の表に示します。

   パラメータ	構成
   データソース	データソース： MySQL データソース名： user_behavior_analysis_mysql
   マイリソースグループ	Prepare the environment で購入した Serverless リソースグループを選択します。
   データ宛先	データ宛先： MaxCompute データソース名： user_behavior_analysis_odps

3. タスクを構成します。

   • ソースと宛先を構成します。

     モジュール	パラメータ	構成
     データソース	テーブル	MySQL テーブル ods_user_info_d を選択します。
     	シャードキー	主キーまたはインデックス列を分割キーとして使用します。整数型フィールドのみサポートされています。 このチュートリアルでは、分割キーを uid フィールドに設定します。
     データ宛先	Tunnel リソースグループ	このチュートリアルでは、デフォルトで common transmission resources を使用します。専用の Tunnel クォータがある場合は、ドロップダウンリストから選択できます。 説明 MaxCompute のデータ転送リソースの詳細については、「Purchase and use exclusive resource groups for data transmission」をご参照ください。支払い遅延または有効期限切れにより専用の Tunnel クォータを使用できない場合、タスクは自動的に common transmission resources を使用するように切り替わります。
     	スキーマ	このチュートリアルではデフォルトスキーマを使用します。MaxCompute プロジェクトに他のスキーマがある場合は、ドロップダウンリストから選択できます。
     	テーブル	事前に作成した ods_user_info_d_odps テーブルを選択します。
     	パーティション情報	このチュートリアルでは、${bizdate} パラメータを入力します。
     	書き込みモード	このチュートリアルでは、[Clean up existing data before writing (Insert Overwrite)] を選択します。 以下の書き込みモードがサポートされています： insert into： テーブルまたは静的パーティションにデータを直接挿入します。 insert overwrite： 新しいデータを挿入する前に、テーブルまたは静的パーティション内の既存データをクリアします。
     	空文字列を Null に変換して書き込み	このチュートリアルでは、いいえ を選択します。

   • フィールドマッピングと一般設定を確認します。

     DataWorks では、ソースフィールドを宛先フィールドにマッピングできます。これにより、指定したソースフィールドのデータが正しい宛先フィールドに書き込まれます。また、読み取り/書き込みの同時実行の設定、同期がデータベースのパフォーマンスに影響しないようにするためのレート制限、ダーティデータ処理の定義、分散タスクの実行などの機能も提供されます。このチュートリアルではデフォルト設定を使用します。その他の構成オプションの詳細については、「Configure a batch synchronization node by using the codeless UI」をご参照ください。

4. スケジューリングプロパティを構成します。

   構成ページで、右側ペインの [Scheduling settings] をクリックして [スケジューリング設定] パネルを開きます。ここで、スケジューリングとノード情報を構成できます。詳細については、「Configure scheduling properties for a node」をご参照ください。このチュートリアルでは、次の構成を使用します。

   パラメータ	構成
   スケジューリングパラメーター	デフォルト値 $bizdate を使用します。 説明 bizdate=$bizdate は、前日の日付を yyyymmdd 形式で取得します。
   時間プロパティ	スケジューリング周期： Day。 スケジュール時刻： 00:30。 再実行プロパティ：前回の成功/失敗に関係なく、タスクの再実行を許可します。 その他のパラメータはデフォルト値を使用します。 説明 ノードは毎日 00:30 以降にのみ実行されます。
   リソースプロパティ	Prepare the environment で作成した Serverless リソースグループを選択します。
   スケジュール依存	依存する先祖ノード を確認します：現在のノードの 親ノード として workshop_start ノードが表示されていることを確認します。ノードを接続して設定した親ノードがここに表示されます。workshop_start ノードが表示されない場合は、「Design a business flow」を参照して、データ同期フェーズの業務フローを設計していることを確認してください。 このユースケースでは、workshop_start ノードがスケジュール時刻に実行され、正常に完了した後に現在のノードがトリガーされます。 ノード出力名が [Name of the MaxCompute production project].ods_user_info_d_odps のノード出力が存在することを確認します。存在しない場合は、この 出力名 で手動で追加する必要があります。 説明 DataWorks はノード出力を使用して依存関係を管理します。自動解析 機能はテーブルリネージを使用し、必要な依存関係を自動的に追加することで、下流の SQL ノードが同期タスクの出力を処理できるようにします。同期待ち出力テーブル [Name of the MaxCompute production project].ods_user_info_d_odps と同じ名前の [ノード出力] が存在するかどうかを確認する必要があります。

ユーザーログ同期パイプラインの構成

このセクションでは、バッチ同期ノードを構成して、提供されているパブリック HttpFile データソースの user_log.txt ファイルからユーザーの Web サイトアクセスデータを MaxCompute テーブル ods_raw_log_d_odps に同期する方法を説明します。

1. ods_raw_log_d_odps バッチ同期ノードをダブルクリックして、ノードエディターを開きます。

2. ネットワークとリソースを構成します。

   データソース、マイリソースグループ、データ宛先 を構成した後、次へ をクリックし、画面の指示に従って接続テストを完了します。必要な構成を以下の表に示します。

   パラメータ	構成
   データソース	データソース： HttpFile データソース名： user_behavior_analysis_httpfile
   マイリソースグループ	Prepare the environment で購入した Serverless リソースグループを選択します。
   データ宛先	データ宛先： MaxCompute データソース名： user_behavior_analysis_odps

3. タスクを構成します。

   • ソースと宛先を構成します。

     モジュール	パラメータ		構成
     データソース	ファイルパス		このチュートリアルでは、/user_log.txt を入力します。
     	テキストタイプ		text を選択します。
     	列区切り文字		このチュートリアルでは、| を入力します。
     	詳細設定	エンコーディング	UTF-8 エンコーディング形式を選択します。
     		圧縮フォーマット	None を選択します。
     		テーブルヘッダーをスキップするかどうか	No を選択します。
     データ宛先	Tunnel リソースグループ		このチュートリアルでは、デフォルトで common transmission resources を使用します。専用の Tunnel クォータがある場合は、ドロップダウンリストから選択できます。 説明 MaxCompute のデータ転送リソースの詳細については、「Purchase and use exclusive resource groups for data transmission」をご参照ください。支払い遅延または有効期限切れにより専用の Tunnel クォータを使用できない場合、タスクは自動的に common transmission resources を使用するように切り替わります。
     	スキーマ		このチュートリアルではデフォルトスキーマを使用します。DataWorks プロジェクトに他のスキーマがある場合は、ドロップダウンリストから選択できます。
     	テーブル		事前に作成した ods_raw_log_d_odps テーブルを選択します。
     	パーティション情報		このチュートリアルでは、${bizdate} パラメータを入力します。
     	書き込みモード		このチュートリアルでは、[Clean up existing data before writing (Insert Overwrite)] を選択します。 以下の書き込みモードがサポートされています： insert into： テーブルまたは静的パーティションにデータを直接挿入します。 insert overwrite： 新しいデータを挿入する前に、テーブルまたは静的パーティション内の既存データをクリアします。
     	空文字列を Null に変換して書き込み		このチュートリアルでは、いいえ を選択します。

     ソースを構成した後、テーブルのデータ構造の確認 をクリックして、ログファイルを正しく読み取れることを確認します。

   • フィールドマッピングと一般設定を確認します。

     DataWorks では、ソースフィールドを宛先フィールドにマッピングできます。これにより、指定したソースフィールドのデータが正しい宛先フィールドに書き込まれます。また、読み取り/書き込みの同時実行の設定、同期がデータベースのパフォーマンスに影響しないようにするためのレート制限、ダーティデータ処理の定義、分散タスクの実行などの機能も提供されます。このチュートリアルではデフォルト設定を使用します。その他の構成オプションの詳細については、「Configure a batch synchronization node by using the codeless UI」をご参照ください。

4. スケジューリングプロパティを構成します。

   構成ページで、右側ペインの [Scheduling settings] をクリックして [スケジューリング設定] パネルを開きます。ここで、スケジューリングとノード情報を構成できます。詳細については、「Configure scheduling properties for a node」をご参照ください。このチュートリアルでは、次の構成を使用します。

   パラメータ	構成
   スケジューリングパラメーター	デフォルト値 $bizdate を使用します。 説明 bizdate=$bizdate は、前日の日付を yyyymmdd 形式で取得します。
   時間プロパティ	スケジューリング周期： Day。 スケジュール時刻： 00:30。 再実行プロパティ：前回の成功/失敗に関係なく、タスクの再実行を許可します。 その他のパラメータはデフォルト値を使用します。 説明 ノードは毎日 00:30 以降にのみ実行されます。
   リソースプロパティ	Prepare the environment で作成した Serverless リソースグループを選択します。
   スケジュール依存	依存する先祖ノード を確認します：現在のノードの 親ノード として workshop_start ノードが表示されていることを確認します。ノードを接続して設定した親ノードがここに表示されます。workshop_start ノードが表示されない場合は、「Design a business flow」で説明されているとおり、業務データ同期フェーズの業務フローを設計していることを確認してください。 このユースケースでは、workshop_start ノードがスケジュール時刻に実行され、正常に完了した後に現在のノードがトリガーされます。 ノード出力名が [Name of the MaxCompute production project].ods_raw_log_d_odps のノード出力が存在することを確認します。存在しない場合は、この 出力名 で手動で追加する必要があります。 説明 DataWorks はノード出力を使用して依存関係を管理します。自動解析 機能はテーブルリネージを使用し、必要な依存関係を自動的に追加することで、下流の SQL ノードが同期タスクの出力を処理できるようにします。同期待ち出力テーブル [Name of the MaxCompute production project].ods_raw_log_d_odps と同じ名前の [ノード出力] が存在するかどうかを確認する必要があります。

ビジネスフローの実行

1. [DataStudio] ページで、業務フロー の下にある User Profile Analysis_MaxCompute をダブルクリックします。 ビジネスフロー パネルが開いたら、ツールバーの アイコンをクリックして、アップストリームおよびダウンストリームの依存関係に基づいてビジネスフローを実行します。

2. 実行を確認します：

   • タスクの実行ステータスを確認します。 状態のノードは、同期が正常に完了したことを意味します。

   • 実行ログの表示: ods_user_info_d_odps ノードと ods_raw_log_d_odps ノードを右クリックし、[ログの表示] を選択します。 ログに以下のキーワードが含まれている場合、ノードが正常に実行され、データが同期されたことを示します。

     正常な実行を示すキーワードには、「completed successfully」、「Shell run successfully!」、「Exit with SUCCESS」などがあります。

同期結果の検証

ビジネスフローが実行されると、MySQL の ods_user_info_d テーブルのデータが workshop2024_01_dev.ods_user_info_d_odps の前日のパーティションに同期され、OSS の user_log.txt ファイルのログが workshop2024_01_dev.ods_raw_log_d_odps の前日のパーティションに同期されます。SQL クエリを本番環境に公開する必要がないため、アドホッククエリを作成して結果を検証できます。

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

   [データ開発] ページの左側のメニューで アイコンをクリックし、[Ad Hoc Query] ペインを開きます。[Ad Hoc Query] を右クリックし、ノードの作成 > [ODPS SQL] を選択します。

2. 宛先テーブルをクエリします。

   次の SQL 文を実行して、データが正しく書き込まれたことを確認します。ods_raw_log_d_odps テーブルと ods_user_info_d_odps テーブルにインポートされたレコード数を確認します。

   // 'your_business_date' を実際の業務日付に置き換えます。たとえば、タスクが 20230621 に実行された場合、業務日付は 20230620 (実行日の前日) です。
   select count(*) from ods_user_info_d_odps  where dt='your_business_date'; 
   select count(*) from ods_raw_log_d_odps where dt='your_business_date';

   同期が成功した場合、ods_user_info_d_odps はゼロ以外のカウント (たとえば 6) を返し、ods_raw_log_d_odps もゼロ以外のカウントを返します。これは、データが対応するパーティションに正常に書き込まれたことを示します。

   説明

   このチュートリアルは DataStudio (開発環境) で実行されるため、データはデフォルトで開発環境の MaxCompute プロジェクト workshop2024_01_dev 内の指定されたテーブルに書き込まれます。