DataWorks:データの同期

MySQL データソース (user_behavior_analysis_mysql) の作成

このチュートリアルの基本的なユーザー情報は MySQL データベースに保存されています。環境を準備したときに作成したプライベート OSS バケットにユーザーデータ (ods_user_info_d) を同期するには、MySQL データソースを作成する必要があります。

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

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

   2. 左側のナビゲーションウィンドウで、[データソース] をクリックしてデータソースページに移動します。

2. [データソースの追加] をクリックします。データソースタイプとして [MySQL] を検索して選択します。

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

   次の表に、主要なパラメーターを示します。その他のパラメーターについては、デフォルト値のままにすることができます。

   パラメーター	説明
   データソース名	データソースの名前を入力します。このチュートリアルでは、user_behavior_analysis_mysql と入力します。
   データソースの説明	このデータソースは DataWorks チュートリアル用です。プラットフォームが提供するテストデータにアクセスするためにオフライン同期タスクを構成する際に、このデータソースからデータを読み取ります。このデータソースはデータ統合シナリオでのみ読み取り可能です。他のモジュールでは使用できません。
   構成方法	[接続文字列] を選択します。
   エンドポイント	ホスト IP アドレス: rm-bp1z69dodhh85z9qa.mysql.rds.aliyuncs.com ポート: 3306
   データベース名	データベース名を入力します。このチュートリアルでは、workshop と入力します。
   ユーザー名	ユーザー名を入力します。このチュートリアルでは、workshop と入力します。
   パスワード	パスワードを入力します。このチュートリアルでは、workshop#2017 と入力します。
   認証オプション	なし。

4. [接続設定] セクションで、本番環境と開発環境の両方について [接続性のテスト] をクリックします。接続ステータスが [接続済み] であることを確認します。

   重要

   • リソースグループがワークスペースにアタッチされ、パブリックネットワークアクセスが有効になっていることを確認してください。そうでない場合、データ同期は失敗します。詳細については、「環境の準備」をご参照ください。

   • 利用可能なリソースグループがない場合は、接続設定セクションのプロンプトに従ってください。[購入に移動] と [購入したリソースグループをアタッチ] をクリックします。

5. [作成を完了] をクリックします。

user_behavior_analysis_httpfile という名前の HttpFile データソースの作成

このチュートリアルのユーザー Web サイトアクセスレコードは OSS に保存されています。環境を準備したときに作成したプライベート OSS バケットにユーザー Web サイトアクセスレコード (user_log.txt) を同期するには、HttpFile データソースを作成する必要があります。

1. [管理センター] ページで、左側のナビゲーションウィンドウにある [データソース] をクリックします。

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

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

   次の表に、主要なパラメーターを示します。その他のパラメーターについては、デフォルト値のままにすることができます。

   パラメーター	説明
   データソース名	データソース名を入力します。このチュートリアルでは、user_behavior_analysis_httpfile と入力します。
   データソースの説明	このデータソースは DataWorks チュートリアル用です。プラットフォームが提供するテストデータにアクセスするためにオフライン同期タスクを構成する際に、このデータソースからデータを読み取ります。このデータソースはデータ統合シナリオでのみ読み取り可能です。他のモジュールでは使用できません。
   URL ドメイン名	開発環境と本番環境の両方で、[URL ドメイン名] を https://dataworks-workshop-2024.oss-cn-shanghai.aliyuncs.com に設定します。

4. [接続設定] セクションで、本番環境と開発環境の両方について [接続性のテスト] をクリックします。接続ステータスが [接続済み] であることを確認します。

   重要

   • リソースグループがワークスペースにアタッチされ、パブリックネットワークアクセスが有効になっていることを確認してください。そうでない場合、データ同期は失敗します。詳細については、「環境の準備」をご参照ください。

   • 利用可能なリソースグループがない場合は、接続設定セクションのプロンプトに従ってください。[購入に移動] と [購入したリソースグループをアタッチ] をクリックします。

5. [作成を完了] をクリックします。

OSS データソースの作成

このチュートリアルでは、MySQL データソースからのユーザー情報と HttpFile データソースからのログ情報が OSS データソースに同期されます。

1. [管理センター] ページで、[データソース] ページに移動し、[データソースの追加] をクリックします。

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

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

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

4. 指定されたリソースグループについて、[接続ステータス (開発環境)] および [接続ステータス (本番環境)] 列で [接続性のテスト] をクリックします。テストが完了するのを待ち、ステータスが [接続済み] であることを確認します。

   説明

   少なくとも 1 つのリソースグループが [接続済み] 状態であることを確認してください。そうでない場合、このデータソースの同期タスクをコードレス UI を使用して作成することはできません。

5. [作成を完了] をクリックして OSS データソースを作成します。

初期ノードの構成

1. ワークフローオーケストレーションページで、workshop_start_spark ノードにカーソルを合わせ、[ノードを開く] をクリックします。

2. workshop_start_spark ノード編集ページの右側のペインで、[スケジューリング設定] をクリックし、パラメーターを構成します。次の表に、主要なパラメーターを示します。記載されていないパラメーターについては、デフォルト値のままにすることができます。

   スケジューリング設定パラメーター	説明
   スケジューリングタイプ	このチュートリアルでは、タイプを ドライランスケジューリング に設定します。
   スケジューリングリソースグループ	このチュートリアルでは、環境を準備したときに作成されたサーバーレスリソースグループを選択します。
   ノード依存関係の構成	workshop_start_spark は初期ノードであるため、上流の依存関係はありません。[ワークスペースのルートノードを使用] をクリックして、ワークスペースのルートノードにワークフローをトリガーさせます。 ワークスペースのルートノードの名前は WorkspaceName_root です。

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

ods_raw_log_d_2oss_spark バッチ同期ノードは、HttpFile データソースからプライベート OSS データソースにユーザーログ情報を同期します。

1. ワークフローオーケストレーションページで、ods_raw_log_d_2oss_spark ノードにカーソルを合わせ、ノードを開くボタンをクリックしてノード構成ページに移動します。

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

   パラメーター	説明
   データソース	データソース: HttpFile。 データソース名: user_behavior_analysis_httpfile。
   マイリソースグループ	環境準備中に購入したサーバーレスリソースグループを選択します。
   データ宛先	データ宛先: OSS。 データソース名: 先に作成したプライベート OSS データソースを選択します。この例では、test_g を選択します。

3. [次へ] をクリックして同期タスクを構成します。

   • データソースと宛先の構成

     次の表に、主要なパラメーターを示します。記載されていないパラメーターについては、デフォルト値のままにすることができます。

     パラメーター	説明
     データソース	ファイルパス: /user_log.txt。 テキストタイプ: text を選択します。 列区切り文字: | を入力します。 圧縮フォーマット: None、Gzip、Bzip2、Zip が含まれます。None を選択します。 ヘッダーをスキップ: いいえを選択します。
     データ宛先	テキストタイプ: text を選択します。 ファイル名 (パスを含む): OSS バケット内のディレクトリに基づいてパスを入力します。例: ods_raw_log_d/log_${bizdate}/log_${bizdate}.txt。ここで、ods_raw_log_d は作成したディレクトリ名、$bizdate は前日の日付を表します。 列区切り文字: | を入力します。

   • [フィールドマッピング] と [チャネルコントロール] を確認します。

     DataWorks は、構成されたフィールドマッピングに基づいて、指定されたソースフィールドから指定された宛先フィールドにデータを同期します。また、同時実行数を設定し、ダーティデータのポリシーを構成することもできます。このチュートリアルでは、[ダーティデータポリシー] を [ダーティデータを許容しない] に設定します。その他の設定はデフォルト値のままにすることができます。詳細については、「コードレス UI を使用して同期タスクを構成する」をご参照ください。

4. デバッグパラメーターを構成します。

   バッチ同期タスク構成ページの右側のペインで、[デバッグ設定] をクリックします。ステップ 6 でデバッグパラメーターを使用してテストを実行するために、次のパラメーターを構成します。

   構成項目	構成の説明
   リソースグループ	環境を準備したときに購入したサーバーレスリソースグループを選択します。
   スクリプトパラメーター	[パラメーターの追加] をクリックし、bizdate に yyyymmdd 形式で値を割り当てます (例: bizdate=20250223)。デバッグ中、DataStudio はタスクで定義された変数をこの定数に置き換えます。

5. (オプション) スケジューリングプロパティを構成します。

   このチュートリアルでは、スケジューリングパラメーターのデフォルト値をそのまま使用できます。ノード編集ページの右側のペインで [スケジューリング設定] をクリックできます。パラメーターの詳細については、「ノードのスケジューリングプロパティを構成する」をご参照ください。

   • スケジューリングパラメーター: これらはこのチュートリアルのワークフローですでに構成されています。内部ノードに対して構成する必要はありません。タスクやコードで直接使用できます。

   • スケジューリングポリシー: [遅延実行時間] パラメーターを設定して、ワークフローの開始後に子ノードが実行を待機する時間を指定できます。このチュートリアルでは、このパラメーターを設定する必要はありません。

6. 上部のツールバーで、[保存] をクリックして現在のノードを保存します。

ユーザーデータ同期パイプライン (ods_user_info_d_2oss_spark) の構成

ods_user_info_d_2oss_Spark バッチ同期ノードは、MySQL データソースからプライベート OSS データソースにユーザーデータを同期します。

1. ワークフローオーケストレーションページで、ods_user_info_d_2oss_Spark ノードにカーソルを合わせ、[ノードを開く] をクリックします。

2. 同期パイプラインのネットワークとリソースを構成します。

   パラメーター	説明
   データソース	データソース: MySQL。 データソース名: user_behavior_analysis_mysql。
   マイリソースグループ	環境準備中に購入したサーバーレスリソースグループを選択します。
   データ宛先	データ宛先: OSS。 データソース名: 先に作成したプライベート OSS データソースを選択します。この例では、test_g を選択します。

3. [次へ] をクリックして同期タスクを構成します。

   • データソースと宛先の構成

     次の表に、主要なパラメーターを示します。記載されていないパラメーターについては、デフォルト値のままにすることができます。

     パラメーター	説明
     データソース	テーブル: データソースから ods_user_info_d を選択します。 シャードキー: プライマリキーまたはインデックス付き列をシャードキーとして使用します。整数フィールドのみがサポートされます。シャードキーを uid に設定します。
     データ宛先	テキストタイプ: text を選択します。 ファイル名 (パスを含む): OSS バケット内のディレクトリに基づいてパスを入力します。例: ods_user_info_d/user_${bizdate}/user_${bizdate}.txt。ここで、ods_user_info_d は作成したディレクトリ名、$bizdate は前日の日付を表します。 列区切り文字: | を入力します。

   • [フィールドマッピング] と [チャネルコントロール] を確認します。

     DataWorks は、構成されたフィールドマッピングに基づいて、指定されたソースフィールドから指定された宛先フィールドにデータを同期します。また、同時実行数を設定し、ダーティデータのポリシーを構成することもできます。このチュートリアルでは、[ダーティデータポリシー] を [ダーティデータを許容しない] に設定します。その他の設定はデフォルト値のままにすることができます。詳細については、「コードレス UI を使用して同期タスクを構成する」をご参照ください。

4. デバッグパラメーターを構成します。

   バッチ同期タスク構成ページの右側のペインで、[デバッグ設定] をクリックします。ステップ 6 でデバッグパラメーターを使用してテストを実行するために、次のパラメーターを構成します。

   構成項目	構成の説明
   リソースグループ	環境を準備したときに購入したサーバーレスリソースグループを選択します。
   スクリプトパラメーター	[パラメーターの追加] をクリックし、bizdate に yyyymmdd 形式で値を割り当てます (例: bizdate=20250223)。デバッグ中、DataStudio はタスクで定義された変数をこの定数に置き換えます。

5. (オプション) スケジューリングプロパティを構成します。

   このチュートリアルでは、スケジューリングパラメーターのデフォルト値をそのまま使用できます。ノード編集ページの右側のペインで [スケジューリング設定] をクリックできます。パラメーターの詳細については、「ノードのスケジューリングプロパティを構成する」をご参照ください。

   • スケジューリングパラメーター: これらはこのチュートリアルのワークフローですでに構成されています。内部ノードに対して構成する必要はありません。タスクやコードで直接使用できます。

   • スケジューリングポリシー: [遅延実行時間] パラメーターを設定して、ワークフローの開始後に子ノードが実行を待機する時間を指定できます。このチュートリアルでは、このパラメーターを設定する必要はありません。

6. 上部のツールバーで、[保存] をクリックして現在のノードを保存します。

ログテーブル (ods_raw_log_d_spark) の作成とデータの解析

データがバッチ統合タスクによってプライベート OSS データソースに同期された後、EMR SPARK SQL ノードを使用して、生成された OSS ファイルに基づいて ods_raw_log_d_spark 外部テーブルを作成できます。LOCATION 句を使用して、バッチデータ統合タスクによってプライベート OSS [バケット] に書き込まれたログ情報にアクセスします。

1. ワークフローオーケストレーションページで、ods_raw_log_d_spark ノードにカーソルを合わせ、[ノードを開く] をクリックします。

2. テーブル作成ステートメントを編集します。

   -- シナリオ: 次の Spark SQL コードは、EMR Spark SQL ノードを使用して ods_raw_log_d_spark 外部テーブルを作成します。
   -- LOCATION を使用して、バッチデータ統合タスクによってプライベート OSS バケットに書き込まれたログ情報を取得し、対応する dt パーティションを追加します。
   -- 注:
   --      DataWorks は、スケジューリングシナリオで宛先テーブルの対応するビジネスパーティションに日次増分データを書き込むためのスケジューリングパラメーターを提供します。
   --      開発シナリオでは、${VariableName} 形式でコード変数を定義できます。
   --      その後、スケジューリング設定ページで、スケジューリングパラメーターを値として変数に割り当てることができます。これにより、スケジューリング中にコードで動的なパラメーター入力が可能になります。
   CREATE EXTERNAL TABLE IF NOT EXISTS ods_raw_log_d_spark
   (
     `col` STRING
   ) 
   PARTITIONED BY (
     dt STRING
   )
   LOCATION 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ods_raw_log_d/log_${bizdate}/';
   
   ALTER TABLE ods_raw_log_d_spark ADD IF NOT EXISTS PARTITION (dt = '${bizdate}') 
   LOCATION 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ods_raw_log_d/log_${bizdate}/'
   

   説明

   必要に応じて、コード内の location アドレスを置き換えてください。dw-spark-demo は、環境を準備したときに作成した OSS バケットの名前です。

3. デバッグパラメーターを構成します。

   [EMR SPARK SQL] タスク構成ページの右側のペインで、[デバッグ設定] をクリックします。ステップ 6 でデバッグパラメーターを使用してテストを実行するために、次のパラメーターを構成します。

   構成項目	構成の説明
   計算リソース	環境を準備したときにアタッチされた Spark 計算リソースを選択します。
   リソースグループ	環境を準備したときに購入したサーバーレスリソースグループを選択します。
   スクリプトパラメーター	[パラメーターの追加] をクリックし、bizdate に yyyymmdd 形式で値を割り当てます (例: bizdate=20250223)。デバッグ中、DataStudio はタスクで定義された変数をこの定数に置き換えます。

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

   このチュートリアルでは、スケジューリングパラメーターのデフォルト値をそのまま使用できます。ノード編集ページの右側のペインで [スケジューリング設定] をクリックできます。パラメーターの詳細については、「ノードのスケジューリングプロパティを構成する」をご参照ください。

   • スケジューリングパラメーター: これらはこのチュートリアルのワークフローですでに構成されています。内部ノードに対して構成する必要はありません。タスクやコードで直接使用できます。

   • スケジューリングポリシー: [遅延実行時間] パラメーターを設定して、ワークフローの開始後に子ノードが実行を待機する時間を指定できます。このチュートリアルでは、このパラメーターを設定する必要はありません。

5. 上部のツールバーで、[保存] をクリックして現在のノードを保存します。

ユーザーテーブル (ods_user_info_d_spark) の作成とデータの解析

データがバッチ統合タスクによってプライベート OSS データソースに同期された後、EMR SPARK SQL ノードを使用して、生成された OSS ファイルに基づいて ods_user_info_d_spark 外部テーブルを作成できます。LOCATION 句を使用して、バッチデータ統合タスクによってプライベート OSS [バケット] に書き込まれたユーザー情報にアクセスします。

1. ワークフローオーケストレーションページで、ods_user_info_d_spark ノードにカーソルを合わせ、[ノードを開く] をクリックします。

2. 同期リンクのネットワークとリソースを構成できます。

   -- シナリオ: 次の Spark SQL コードは、EMR Spark SQL ノードを使用して ods_user_info_d_spark 外部テーブルを作成します。
   -- LOCATION を使用して、バッチデータ統合タスクによってプライベート OSS バケットに書き込まれたユーザー情報を取得し、対応する dt パーティションに書き込みます。
   -- 注:
   --      DataWorks は、スケジューリングシナリオで宛先テーブルの対応するビジネスパーティションに日次増分データを書き込むためのスケジューリングパラメーターを提供します。
   --      開発シナリオでは、${VariableName} 形式でコード変数を定義できます。
   --      その後、スケジューリング設定ページで、スケジューリングパラメーターを値として変数に割り当てることができます。これにより、スケジューリング中にコードで動的なパラメーター入力が可能になります。
   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-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ods_user_info_d/user_${bizdate}/'
   ;
   
   ALTER TABLE ods_user_info_d_spark ADD IF NOT EXISTS PARTITION (dt = '${bizdate}') 
   LOCATION'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ods_user_info_d/user_${bizdate}/'
   ;

   説明

   必要に応じて、コード内の location アドレスを置き換えてください。dw-spark-demo は、環境を準備したときに作成した OSS バケットの名前です。

3. デバッグパラメーターを構成します。

   [EMR SPARK SQL] タスク構成ページの右側のペインで、[デバッグ設定] をクリックします。ステップ 6 でデバッグパラメーターを使用してテストを実行するために、次のパラメーターを構成します。

   構成項目	構成の説明
   計算リソース	環境を準備したときにアタッチされた Spark 計算リソースを選択します。
   リソースグループ	環境を準備したときに購入したサーバーレスリソースグループを選択します。
   スクリプトパラメーター	[パラメーターの追加] をクリックし、bizdate に yyyymmdd 形式で値を割り当てます (例: bizdate=20250223)。デバッグ中、DataStudio はタスクで定義された変数をこの定数に置き換えます。

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

   このチュートリアルでは、スケジューリングパラメーターのデフォルト値をそのまま使用できます。ノード編集ページの右側のペインで [スケジューリング設定] をクリックできます。パラメーターの詳細については、「ノードのスケジューリングプロパティを構成する」をご参照ください。

   • スケジューリングパラメーター: これらはこのチュートリアルのワークフローですでに構成されています。内部ノードに対して構成する必要はありません。タスクやコードで直接使用できます。

   • スケジューリングポリシー: [遅延実行時間] パラメーターを設定して、ワークフローの開始後に子ノードが実行を待機する時間を指定できます。このチュートリアルでは、このパラメーターを設定する必要はありません。

5. 上部のツールバーで、[保存] をクリックして現在のノードを保存します。