E-MapReduce:データの同期

ユーザーの Web サイトアクセスログを非公開 OSS バケットに同期するようにノードを構成する

バッチ同期タスクを使用して、HttpFile データソースに格納されているユーザーの Web サイトアクセスログを非公開 OSS バケットに同期できます。

HttpFile データソースの Web サイトアクセスログを作成した OSS バケットに同期します。

1. [DataStudio] ページで、ods_raw_log_d_2oss_spark ノードをダブルクリックして、ノードの構成タブに移動します。

2. 使用するリソースグループとデータソースの間にネットワーク接続を確立します。

   ネットワーク接続とリソースの構成が完了したら、[次へ] をクリックし、プロンプトに従って接続テストを完了します。

   パラメーター	説明
   [ソース]	ソース: 値を HttpFile に設定します。 データソース名: 値を user_behavior_analysis_httpfile に設定します。
   [リソースグループ]	購入した [サーバーレスリソースグループ] を選択します。
   [デスティネーション]	デスティネーション: 値を OSS に設定します。 データソース名: 追加した非公開 OSS データソースの名前を指定する test_g に設定します。

3. データ同期ノードを構成します。

   パラメーター	説明
   ソース	[ファイルパス]: 値を /user_log.txt に設定します。 [テキストタイプ]: 値を text に設定します。 [列区切り文字]: | を入力します。 [圧縮形式]: OSS オブジェクトの圧縮形式。有効な値: なし、Gzip、Bzip2、および Zip。 なし を選択します。 [ヘッダーをスキップ]: 値を いいえ に設定します。
   [デスティネーション]	[テキストタイプ]: 値を text に設定します。 [オブジェクト名 (パスを含む)]: OSS オブジェクトのパス。OSS バケットに作成したフォルダーに基づいてこのパラメーターを構成します。この例では、ods_raw_log_d/log_${bizdate}/log_${bizdate}.txt が入力されています。 ods_raw_log_d は、OSS バケットに作成したフォルダーの名前です。 $bizdate は前日の日付を示します。 [列区切り文字]: | を入力します。

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

   ノードの構成タブの右側のナビゲーションウィンドウで、[プロパティ] をクリックします。 [プロパティ] タブで、ノードのスケジューリングプロパティと基本情報を構成します。次の表に、構成を示します。

   説明

   DataWorks は、スケジューリングシナリオで日次データを OSS バケットのさまざまなディレクトリとオブジェクトに書き込むために使用できるスケジューリングパラメーターを提供します。ディレクトリとオブジェクトには、データタイムスタンプに基づいて名前が付けられます。

   ビジネスシナリオに基づいて、非公開 OSS バケットの [オブジェクト名 (パスを含む)] パラメーターを構成するときに、${変数名} 形式でディレクトリに変数を構成できます。次に、ノードの構成タブの [プロパティ] タブで、変数にスケジューリングパラメーターを値として割り当てることができます。このようにして、非公開 OSS バケット内のディレクトリ名とオブジェクト名を、スケジューリングシナリオで動的に生成できます。

   セクション	説明	スクリーンショット
   [スケジューリングパラメーター]	このセクションで、[パラメーターを追加] をクリックしてスケジューリングパラメーターを追加します。 パラメーター名を bizdate に設定します。 パラメーター値を $[yyyymmdd-1] に設定します。 詳細については、「スケジューリングパラメーターを構成する」をご参照ください。
   [依存関係]	このセクションで、出力テーブルが現在のノードの出力名として使用されていることを確認します。 出力テーブルには、ワークスペース名.ノード名 の形式で名前が付けられます。 詳細については、「スケジューリングの依存関係を構成する」をご参照ください。

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

ユーザーの基本情報を非公開 OSS バケットに同期するようにノードを構成する

バッチ同期タスクを使用して、MySQL データソースに格納されているユーザーの基本情報を非公開 OSS バケットに同期できます。

ods_raw_log_d_spark ノードを構成する

EMR Spark SQL ノードを使用して作成された ods_raw_log_d_spark 外部テーブルに基づいて、LOCATION を使用して、関連するバッチ同期タスクによって [非公開 OSS] [バケット] に同期されたユーザーの Web サイトアクセスログにアクセスできます。

1. コードを構成します。

   -- シナリオ: 次のサンプルコードでは、Spark SQL の SQL ステートメントが使用されています。コードでは、ods_raw_log_d_spark 外部テーブルが EMR Spark SQL ノードを使用して作成され、LOCATION を使用して、関連するバッチ同期タスクによって非公開 OSS バケットに同期されたユーザーの Web サイトアクセスログにアクセスします。
   -- 注:
   --      DataWorks は、スケジューリングシナリオで日次の増分データをデスティネーションテーブルの対応するパーティションに書き込むために使用できるスケジューリングパラメーターを提供します。
   --      実際の開発シナリオでは、ノードコードで ${変数名} 形式で変数を定義できます。次に、ノードの構成タブの [プロパティ] タブで、変数にスケジューリングパラメーターを値として割り当てることができます。このようにして、スケジューリングパラメーターの値は、スケジューリングシナリオでのスケジューリングパラメーターの構成に基づいて、ノードコードで動的に置き換えられます。
   CREATE EXTERNAL TABLE IF NOT EXISTS ods_raw_log_d_spark
   (
     `col` STRING
   )
   PARTITIONED BY (
     dt STRING
   )
   LOCATION 'oss://dw-emr-demo/ods_raw_log_d/log_${bizdate}/';
   
   ALTER TABLE ods_raw_log_d_spark ADD IF NOT EXISTS PARTITION (dt = '${bizdate}')
   LOCATION 'oss://dw-emr-demo/ods_raw_log_d/log_${bizdate}/'
   ;

   説明

   上記のコードでは、LOCATION にサンプルパスが使用されています。パスは、関連するバッチ同期ノードを構成するときの [オブジェクト名 (パスを含む)] パラメーターの値に基づいています。 LOCATION パラメーターを作成したフォルダーのパスに設定する必要があります。 dw-emr-demo は、環境を準備したときに作成した OSS バケットのドメイン名です。

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

   ods_raw_log_d_spark ノードのスケジューリングパラメーターを構成して、特定のデータタイムスタンプを持つ非公開 OSS ログオブジェクトを取得し、Spark テーブルの同じデータタイムスタンプを持つ対応するパーティションにデータを書き込みます。

   セクション	説明	スクリーンショット
   [スケジューリングパラメーター]	このセクションで、[パラメーターを追加] をクリックしてスケジューリングパラメーターを追加します。 パラメーター名を bizdate に設定します。 パラメーター値を $[yyyymmdd-1] に設定します。詳細については、「スケジューリングパラメーターを構成する」をご参照ください。
   [依存関係]	このセクションで、出力テーブルが現在のノードの出力名として使用されていることを確認します。 出力テーブルには、ワークスペース名.ノード名 の形式で名前が付けられます。 詳細については、「スケジューリングの依存関係を構成する」をご参照ください。

   説明

   このトピックの SQL ステートメントでは、${bizdate} スケジューリングパラメーターが構成され、値 T-1 がスケジューリングパラメーターに割り当てられます。バッチコンピューティングシナリオでは、bizdate はビジネストランザクションが実行された日付を示し、これは多くの場合データタイムスタンプと呼ばれます。たとえば、現在の日に前日の売上高の統計データを収集する場合、前日はビジネストランザクションが実行された日付であり、データタイムスタンプを表します。

3. 構成が完了したら、 アイコンをクリックします。

ods_user_info_d_spark ノードを構成する

EMR Spark SQL ノードを使用して作成された ods_user_info_d_spark 外部テーブルに基づいて、LOCATION を使用して、関連するバッチ同期タスクによって [非公開 OSS] [バケット] に同期されたユーザーの基本情報にアクセスできます。

1. コードを構成します。

   -- シナリオ: 次のサンプルコードでは、Spark SQL の SQL ステートメントが使用されています。コードでは、ods_user_info_d_spark 外部テーブルが EMR Spark SQL ノードを使用して作成され、LOCATION を使用して、関連するバッチ同期タスクによって非公開 OSS バケットに同期されたユーザーの基本情報にアクセスします。次に、データは dt パーティションに書き込まれます。
   -- 注:
   --      DataWorks は、スケジューリングシナリオで日次の増分データをデスティネーションテーブルの対応するパーティションに書き込むために使用できるスケジューリングパラメーターを提供します。
   --      実際の開発シナリオでは、ノードコードで ${変数名} 形式で変数を定義できます。次に、ノードの構成タブの [プロパティ] タブで、変数にスケジューリングパラメーターを値として割り当てることができます。このようにして、スケジューリングパラメーターの値は、スケジューリングシナリオでのスケジューリングパラメーターの構成に基づいて、ノードコードで動的に置き換えられます。
   CREATE EXTERNAL TABLE IF NOT EXISTS ods_user_info_d_spark
   (
       `uid`        STRING COMMENT 'ユーザー ID'
       ,`gender`    STRING COMMENT '性別'
       ,`age_range` STRING COMMENT '年齢層'
       ,`zodiac`    STRING COMMENT '星座'
   )
   PARTITIONED BY
   (
       dt           STRING
   )
   ROW FORMAT DELIMITED
   FIELDS
   TERMINATED
   BY'|'
   STORED AS TEXTFILE
   LOCATION 'oss://dw-emr-demo/ods_user_info_d/user_${bizdate}/'
   ;
   
   ALTER TABLE ods_user_info_d_spark ADD IF NOT EXISTS PARTITION (dt = '${bizdate}')
   LOCATION'oss://dw-emr-demo/ods_user_info_d/user_${bizdate}/'
   ;

   説明

   上記のコードでは、LOCATION にサンプルパスが使用されています。パスは、関連するバッチ同期ノードを構成するときの [オブジェクト名 (パスを含む)] パラメーターの値に基づいています。 LOCATION パラメーターを作成したフォルダーのパスに設定する必要があります。 dw-emr-demo は、環境を準備したときに作成した OSS バケットのドメイン名です。

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

   ods_user_info_d_spark ノードのスケジューリングパラメーターを構成して、特定のデータタイムスタンプを持つ非公開 OSS ユーザー情報オブジェクトを取得し、Spark テーブルの同じデータタイムスタンプを持つ対応するパーティションにデータを書き込みます。

   セクション	説明	スクリーンショット
   [スケジューリングパラメーター]	このセクションで、[パラメーターを追加] をクリックしてスケジューリングパラメーターを追加します。 パラメーター名を bizdate に設定します。 パラメーター値を $[yyyymmdd-1] に設定します。詳細については、「スケジューリングパラメーターを構成する」をご参照ください。
   [依存関係]	このセクションで、出力テーブルが現在のノードの出力名として使用されていることを確認します。 出力テーブルには、ワークスペース名.ノード名 の形式で名前が付けられます。 詳細については、「スケジューリングの依存関係を構成する」をご参照ください。

3. 構成が完了したら、 アイコンをクリックします。