DataWorks:データの同期

ログ

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

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

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

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

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

   パラメーター	説明
   ソース	ソース: 値を HttpFile に設定します。 データソース名: 値を user_behavior_analysis_httpfile に設定します。
   [リソースグループ]	購入した serverless リソースグループ を選択します。
   [デスティネーション]	デスティネーション: 値を 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. 設定が完了したら、ツールバーの アイコンをクリックします。

ユーザーデータ

バッチ同期タスクを使用して、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. 設定が完了したら、 アイコンをクリックします。