DataWorks:データ同期

HttpFile データソースの作成

1. [データソース] ページに移動します。

   1. DataWorks コンソールにログインします。 左側のナビゲーションウィンドウで、[Management Center] をクリックします。 表示されるページで、ドロップダウンリストから目的のワークスペースを選択し、[管理センターに移動] をクリックします。

   2. 表示されるページの左側のナビゲーションウィンドウで、[データソース] をクリックします。 [データソース] ページが表示されます。

2. Add Connection をクリックします。 Add Connection ダイアログボックスで、[HttpFile] を検索して選択します。

3. [HttpFile データソースの作成] ページで、パラメーターを設定します。 このチュートリアルでは、開発環境と本番環境の両方でサンプル値を使用します。

   パラメーター	説明
   [Data Source Name]	データソースの名前を入力します。 このチュートリアルでは、user_behavior_analysis_httpfile と入力します。
   [Description]	説明を入力します。 このチュートリアル固有のデータソースは、オフライン同期タスク用のテストデータを提供します。 読み取り専用で、データ統合にのみ使用できます。
   URL ドメイン	開発環境と本番環境の両方の URL ドメインは https://dataworks-workshop-2024.oss-cn-shanghai.aliyuncs.com です。

4. リソースグループでは、コネクテッドステート (開発環境) と 接続ステータス (本番環境) の両方の列で テスト接続 をクリックします。 接続ステータスが 接続可能 になるまで待ちます。

   重要

   少なくとも 1 つのリソースグループのステータスが 接続可能 であることを確認してください。 そうでない場合、ウィザードを使用してこのデータソースの同期タスクを作成することはできません。

5. Complete Creation をクリックします。

MySQL データソースの作成

1. Management Center ページで、Data Source ページに移動し、Add Connection をクリックします。

2. [データソースの追加] ダイアログボックスで、MySQL を検索して選択します。

3. [MySQL データソースの作成] ページで、パラメーターを設定します。 このチュートリアルでは、開発環境と本番環境の両方でサンプル値を使用します。

   パラメーター	説明
   [Data Source Name]	データソースの名前を入力します。 このチュートリアルでは、user_behavior_analysis_mysql と入力します。
   [Description]	説明を入力します。 このチュートリアル固有のデータソースは、オフライン同期タスク用のテストデータを提供します。 読み取り専用で、データ統合にのみ使用できます。
   [設定モード]	User-created Data Store with Public IP Addresses を選択します。
   [Connection Address]	ホストアドレス IP： rm-bp1z69dodhh85z9qa.mysql.rds.aliyuncs.com ポート： 3306
   [Database Name]	データベース名を入力します。 このチュートリアルでは、workshop と入力します。
   [Username]	ユーザー名を入力します。 このチュートリアルでは、workshop と入力します。
   [Password]	パスワードを入力します。 このチュートリアルでは、workshop#2017 と入力します。
   [Authentication Method]	認証なし。

4. リソースグループでは、コネクテッドステート (開発環境) と 接続ステータス (本番環境) の両方の列で テスト接続 をクリックします。 接続ステータスが 接続可能 になるまで待ちます。

5. Complete Creation をクリックします。

OSS データソースの作成

このユースケースでは、MySQL データソースからのユーザー情報と HttpFile データソースからのログデータを OSS データソースに同期します。

1. Management Center ページで、Data Source ページに移動し、Add Connection をクリックします。

2. Add Connection ダイアログボックスで、OSS を検索して選択します。

3. [OSS データソースの作成] ページで、パラメーターを設定します。 このチュートリアルでは、開発環境と本番環境の両方でサンプル値を使用します。

   パラメーター	説明
   [Data Source Name]	データソースの名前を入力します。 この例では、test_g を使用します。
   [Description]	データソースの簡単な説明を入力します。
   [Access mode]	Access Key mode を選択します。
   AccessKey ID	現在のアカウントの AccessKey ID です。 AccessKey ページに移動して AccessKey ID をコピーしてください。
   AccessKey シークレット	アカウントの AccessKey シークレットを入力します。 重要 AccessKey シークレットは作成時にのみ表示され、後で表示することはできません。 AccessKey シークレットは機密情報として保管してください。 AccessKey が漏洩または紛失した場合は、それを削除して新しいものを作成してください。
   [エンドポイント]	http://oss-cn-shanghai-internal.aliyuncs.com と入力します。
   バケット	環境を準備する際に EMR クラスター用に設定した OSS バケットの名前を入力します。 このチュートリアルでは、dw-emr-demo を使用します。

4. リソースグループでは、コネクテッドステート (開発環境) と 接続ステータス (本番環境) の両方の列で テスト接続 をクリックします。 接続ステータスが 接続可能 になるまで待ちます。

   説明

   少なくとも 1 つのリソースグループのステータスが 接続可能 であることを確認してください。 そうでない場合、ウィザードを使用してこのデータソースの同期タスクを作成することはできません。

5. Complete Creation をクリックします。

1. 左上隅の アイコンをクリックし、[すべての製品] > データ開発と О&М > DataStudio (データ開発) を選択します。

2. DataStudio ペインで、ビジネスプロセス を右クリックし、Create Workflow を選択します。

3. Create Workflow ダイアログボックスで、Workflow Name に workshop_emr と入力し、New をクリックします。

4. ワークフロー開発キャンバスで、ゼロロードノードと 2 つのオフライン同期ノードを作成します。

   1. New Node をクリックし、Zero-Load Node を右側のキャンバスにドラッグします。 New Node ダイアログボックスで、Node Name に workshop_start_emr と入力し、Confirm をクリックします。

   2. New Node をクリックし、Offline synchronization ノードを選択して右側の編集ページにドラッグします。 MySQL のユーザー情報と HttpFile のログ情報を同期するために、Node Name をそれぞれ ods_raw_log_d_2oss_emr と ods_user_info_d_2oss_emr に設定し、Confirm をクリックします。

5. ワークフロー開発キャンバスで、線をドラッグして workshop_start_emr ノードを 2 つのオフライン同期ノードの上流ノードとして設定します。

初期ノードの設定

1. DataStudio で、ワークフロー内のゼロロードノードをダブルクリックします。 ノードエディター タブで、右側のペインにある Scheduling Configuration をクリックします。

2. スケジューリングプロパティを設定します。

   スケジューリングプロパティ	説明
   スケジュール時間	ゼロロードノードのスケジュール時間を 00:30 に設定します。 ノードは毎日 00:30 に現在のワークフローを開始して実行します。 RUN Attribute を [成功または失敗に関わらず再実行] に設定します。
   スケジューリング依存関係	ゼロロードノード workshop_start_emr には上流の依存関係がないため、その依存関係を Workspace Root Node に設定できます。 ルートノードが workshop_start_emr ノードの実行をトリガーします。 ワークスペースルートノードは WorkspaceName_root という形式で命名されます。

3. 設定が完了したら、上部のツールバーにある アイコンをクリックします。

オフライン同期ノードの設定

1. MySQL データソースから OSS バケットにユーザー情報を同期します。

   1. DataStudio で、ods_user_info_d_2oss_emr ノードをダブルクリックして設定タブを開きます。

   2. ソース、宛先、リソースグループを設定します。

      以下の設定を完了した後、次のステップ をクリックし、プロンプトに従って接続テストを完了してください。

      パラメーター	説明
      [Data source]	タイプ：MySQL。 データソース名：user_behavior_analysis_mysql。
      [My Resource Group]	購入した サーバーレスリソースグループ を選択します。
      [Data going]	タイプ：OSS。 データソース名：作成したプライベート OSS データソースを選択します。 この例では、test_g を使用します。

   3. 同期タスクを設定します。

      パラメーター	説明
      [Data source]	[Table]：データソースから ods_user_info_d を選択します。 [Shard Key]：主キーまたはインデックス付きの列をシャードキーとして使用することを推奨します。 整数型のフィールドのみがサポートされています。 このチュートリアルでは、シャードキーを uid に設定します。
      [Data going]	[File Type]：text を選択します。 [Object Name (Path Included)]：OSS ディレクトリに基づいてパスを入力します。 このチュートリアルでは、ods_user_info_d/user_${bizdate}/user_${bizdate}.txt と入力します。 このパスでは、ods_user_info_d はカスタムディレクトリ名であり、${bizdate} は前日の日付の変数です。 [Field Delimiter]：| と入力します。

   4. スケジュールを設定します。

      右側のペインにある [スケジューリング設定] をクリックします。 Scheduling Configuration ペインで、以下のようにスケジューリングプロパティを設定します。

      パラメーター	設定
      [Scheduling Parameters]	Scheduling Parameters セクションで、次のパラメーターを追加します： パラメーター名：bizdate 値：$[yyyymmdd-1]
      [Time attribute]	RUN Attribute を [成功または失敗に関わらず再実行] に設定します。
      [Scheduling Dependency]	Scheduling Dependency セクションで、テーブルがこのノードの出力として設定されていることを確認してください。 形式は WorkspaceName.NodeName です。

   5. 設定が完了したら、ツールバーの アイコンをクリックします。

2. HttpFile データソースから OSS バケットにログデータを同期します。

   1. DataStudio で、ods_raw_log_d_2oss_emr ノードをダブルクリックして設定タブを開きます。

   2. ソース、宛先、リソースグループを設定します。

      以下の設定を完了した後、次のステップ をクリックし、プロンプトに従って接続テストを完了してください。

      パラメーター	説明
      [Data source]	タイプ：HttpFile。 データソース名：user_behavior_analysis_httpfile。
      [My Resource Group]	購入した サーバーレスリソースグループ を選択します。
      [Data going]	タイプ：OSS。 データソース名：作成したプライベート OSS データソースを選択します。 この例では、test_g を使用します。

   3. 同期タスクを設定します。

      パラメーター	説明
      [Data source]	[File Path]：/user_log.txt。 [File Type]：text を選択します。 [Field Delimiter]：| と入力します。 [Compression Format]：オプションには、None、Gzip、Bzip2、Zip があります。 このチュートリアルでは、[None] を選択します。 [Skip Header]：[いいえ] を選択します。
      [Data going]	[File Type]：text を選択します。 [Object Name (Path Included)]：OSS ディレクトリに基づいてパスを入力します。 このチュートリアルでは、ods_raw_log_d/log_${bizdate}/log_${bizdate}.txt と入力します。 このパスでは、ods_raw_log_d はカスタムディレクトリ名であり、${bizdate} は前日の日付の変数です。 [Field Delimiter]：| と入力します。

   4. スケジュールを設定します。

      右側のペインにある Scheduling Configuration をクリックします。 Scheduling Configuration ペインで、以下のようにスケジューリングプロパティを設定します。

      パラメーター	設定
      [Scheduling Parameters]	Scheduling Parameters セクションで、次のパラメーターを追加します： パラメーター名：bizdate 値：$[yyyymmdd-1]
      [Time attribute]	RUN Attribute を [成功または失敗に関わらず再実行] に設定します。
      [Scheduling Dependency]	Scheduling Dependency セクションで、テーブルがこのノードの出力として設定されていることを確認してください。 形式は WorkspaceName.NodeName です。

   5. 設定が完了したら、ツールバーの アイコンをクリックします。

EMR テーブルの作成

ods_user_info_d_emr と ods_raw_log_d_emr の 2 つのテーブルを作成し、それぞれ同期された生の MySQL ユーザーデータと OSS ログデータをクエリします。

1. DataStudio で、作成したワークフローを開きます。 EMR を右クリックし、New Node > EMR Hive を選択します。

2. New Node ダイアログボックスで、Node Name を入力し、Confirm をクリックします。

   対応するテーブルを作成するために、ods_user_info_d_emr と ods_raw_log_d_emr の 2 つの EMR Hive ノードを作成する必要があります。 ワークフロー開発キャンバスで、線をドラッグしてノードの依存関係を設定します。

   [workshop_start_emr] を [ods_user_info_d_2oss_emr] と [ods_raw_log_d_2oss_emr] の上流ノードとして設定します。 次に、これら 2 つの同期ノードを、それぞれ [ods_user_info_d_emr] と [ods_raw_log_d_emr] の上流ノードとして設定します。

3. それぞれの EMR Hive ノードにテーブル作成ステートメントを入力してください。 Scheduling Configuration で、Scheduling Resource Groups を購入したサーバーレスリソースグループに設定してください。 次に、Save と Run をクリックして、各テーブル作成ステートメントを実行してください。

   • ods_user_info_d_emr テーブルの作成

     ods_user_info_d_emr ノードをダブルクリックしてエディター タブを開き、テーブル作成タスクを編集します。

     1. SQL コードの編集

        CREATE EXTERNAL TABLE IF NOT EXISTS ods_user_info_d_emr
        (
            `uid` STRING COMMENT 'ユーザー ID',
            `gender` STRING COMMENT '性別',
            `age_range` STRING COMMENT '年齢範囲',
            `zodiac` STRING COMMENT '星座'
        ) PARTITIONED BY (
          dt STRING
        )
        ROW FORMAT  delimited fields terminated by '|'
        LOCATION 'oss://dw-emr-demo/ods_user_info_d/';
        
        ALTER TABLE ods_user_info_d_emr ADD IF NOT EXISTS PARTITION (dt='${bizdate}')
        LOCATION 'oss://dw-emr-demo/ods_user_info_d/user_${bizdate}/';

        説明

        コード内の LOCATION はサンプルパスであり、オフライン同期タスクで設定した宛先パスと一致させる必要があります。 実際のファイルパスに置き換えてください。 パス内の dw-emr-demo は、環境準備時に作成した OSS バケットの名前です。

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

        パラメーター	設定
        [Scheduling Parameters]	Scheduling Parameters セクションで、次のパラメーターを追加します： パラメーター名：bizdate 値：$[yyyymmdd-1]
        [Time attribute]	RUN Attribute を [成功または失敗に関わらず再実行] に設定します。
        [Scheduling Dependency]	Scheduling Dependency セクションで、テーブルがこのノードの出力として設定されていることを確認してください。 形式は WorkspaceName.NodeName です。

     3. 設定が完了したら、 をクリックしてノードを保存してください。

   • ods_raw_log_d_emr テーブルの作成

     ods_raw_log_d_emr ノードをダブルクリックし、タスクを編集して ods_raw_log_d_emr テーブルを作成します。

     1. SQL コードの編集

        -- OSSログデータのターゲットテーブルを作成します。
        CREATE EXTERNAL TABLE IF NOT EXISTS ods_raw_log_d_emr
        (
          `col` STRING
        ) PARTITIONED BY (
          dt STRING
        );
        ALTER TABLE ods_raw_log_d_emr ADD IF NOT EXISTS PARTITION (dt='${bizdate}')
        LOCATION 'oss://dw-emr-demo/ods_raw_log_d/log_${bizdate}/';

        説明

        コード内の LOCATION はサンプルパスであり、オフライン同期タスクで設定した宛先パスと一致させる必要があります。 実際のファイルパスに置き換えてください。 パス内の dw-emr-demo は、環境準備時に作成した OSS バケットの名前です。

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

        パラメーター	設定
        [Scheduling Parameters]	Scheduling Parameters セクションで、次のパラメーターを追加します： パラメーター名：bizdate 値：$[yyyymmdd-1]
        [Time attribute]	RUN Attribute を [成功または失敗に関わらず再実行] に設定します。
        [Scheduling Dependency]	Scheduling Dependency セクションで、テーブルがこのノードの出力として設定されていることを確認してください。 形式は WorkspaceName.NodeName です。

     3. 設定が完了したら、 をクリックしてノードを保存してください。

ワークフローの実行

1. [データ開発] ページで、ビジネスプロセス の下にある workshop_emr をダブルクリックし、開いたワークフロー パネルでツールバーの アイコンをクリックして、上流と下流の依存関係に基づいてワークフローを実行してください。

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

   • タスクの実行ステータスを確認します：

     • ノード上の アイコンは、タスクが正常に実行されたことを示します。

     • アイコンと java.net.ConnectException: Connection timed out (Connection timed out) エラーは、ネットワーク接続の問題を示します。 これを解決するには、ECS コンソールで、リソースグループの VPC の vSwitch CIDR ブロックからポート 10000 へのインバウンドトラフィックを許可するセキュリティグループのルールを追加してください。 vSwitch CIDR ブロックを見つけるには、リソースグループ リストに移動し、関連するリソースグループの ネットワーク設定 をクリックします。 セキュリティグループのルールの追加方法の詳細については、「セキュリティグループルールの追加」をご参照ください。

   • 実行ログを表示するには、ods_user_info_d_emr および ods_raw_log_d_emr ノードを右クリックし、[ログの表示] を選択します。 以下のログ出力は、データが正常に同期されたことを示します。

     State:3(RUN) China Standard Time 2019
     Logview: China Standard Time 2019
     
     State:4(SUCCESS) China Standard Time 2019
     
     Total:20028R  442.8KB
     Exit SUCCESS
     
     INFO  Shell run successfully!
     Current task status: FINISH
     Cost Time: 48.309s

同期結果のクエリ

1. アドホッククエリ ノードを作成します。

   DataStudio の左側メニューで、 をクリックしてアドホッククエリ ペインに移動します。 [アドホッククエリ] を右クリックし、New Node>EMR Hive を選択します。

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

   次の SQL ステートメントを実行して、データが ods_raw_log_d_emr テーブルと ods_user_info_d_emr テーブルに書き込まれたことを確認してください。

   -- クエリのパーティション列は業務日に更新する必要があります。 
   -- たとえば、タスクが 20191107 に実行される場合、業務日は 20191106 (タスク実行日の前日) になります。
   SELECT * from ods_user_info_d_emr where dt='your_business_date'; 
   SELECT * from ods_raw_log_d_emr where dt='your_business_date';