DataWorks:データ同期

ワークフローの作成および設計

DataWorks における開発は、ワークフロー単位で管理されます。ノードを追加する前に、まずワークフローを作成します。

1. DataStudio に移動します。DataWorks コンソールにログインします。上部ナビゲーションバーで対象リージョンを選択します。左側ナビゲーションウィンドウで、データ開発および運用 > データ開発 を選択します。ドロップダウンリストからワークスペースを選択し、データ開発へ移動 をクリックします。

2. User profile analysis_MaxCompute という名前のワークフローを作成します。手順については、「ワークフローの作成」をご参照ください。

   

3. ワークフローの設計を行います。ワークフローキャンバスが開いたら、ツールバーの ノードの作成 をクリックします。ノードをキャンバス上にドラッグし、線を引いてデータ同期のスケジューリング依存関係を設定します。レイアウトは「ワークフロー設計」に基づきます。本チュートリアルでは、以下の 3 つのノードを使用します。ゼロロードノードと同期ノード間にはデータリネージがないため、キャンバス上で線を引くことで依存関係を構成します。詳細については、「スケジューリング依存関係の構成ガイド」をご参照ください。

   ノードタイプ	ノード名	目的
   ゼロロードノード	workshop_start_odps	毎日 00:30 にワークフロー全体をトリガーします。このノードにはコードは不要です。
   バッチ同期	ods_user_info_d_odps	MySQL から基本的なユーザー情報を MaxCompute へ同期します。
   バッチ同期	ods_raw_log_d_odps	OSS から Web サイトアクセスログを MaxCompute へ同期します。

   

スケジューリングロジックの構成

workshop_start_odps ゼロロードノードがワークフロー全体をトリガーします。このノードを毎日 00:30 に実行するよう構成し、完了後に同期ノードが自動的に実行されるようにします。他のノードのスケジューリング設定はデフォルトのままとします。

構成項目	設定値	説明
スケジューリング時刻		00:30 に設定します。ゼロロードノードは、この時刻に毎日ワークフローをトリガーします。
スケジューリング依存関係		workshop_start_odps ノードには先祖ノードがありません。ワークスペースのルートノードに依存するよう設定することで、このゼロロードノードをトリガーします。

スケジューリング構成の詳細については、「異なるシナリオにおけるワークフロー内ノードのスケジューリング時刻の構成」および「概要」をご参照ください。

MySQL から MaxCompute への同期

このタスクは、MySQL テーブル ods_user_info_d から読み取り、MaxCompute テーブル ods_user_info_d_odps へ書き込みます。

1. ods_user_info_d_odps ノードをダブルクリックして、構成タブを開きます。

2. ソース、リソースグループ、送信先を構成し、次へ をクリックして接続性テストを完了します。

   パラメーター	値
   ソース	MySQL — データソース名：user_behavior_analysis_mysql
   リソースグループ	ご利用のサーバーレスリソースグループ
   送信先	MaxCompute — データソース名：user_behavior_analysis_mysql

   

3. ソースおよび送信先の設定を構成します。デフォルトのフィールドマッピングおよびチャネル制御設定を使用します。利用可能なすべてのオプションについては、「コードレス UI を使用したバッチ同期タスクの構成」をご参照ください。

   項目	パラメーター	値	スクリーンショット
   ソース	テーブル	ods_user_info_d
   	分割キー	uid
   送信先	Tunnel リソースグループ	共通伝送リソース（デフォルト）
   	スキーマ	default
   	テーブル	ods_user_info_d_odps
   	パーティション情報	${bizdate}
   	書き込みモード	上書き挿入（書き込み前に既存のデータをクリア）
   	空文字列を NULL に変換して書き込み	いいえ

4. スケジューリングプロパティを構成します。右側ナビゲーションペインで プロパティ をクリックし、以下の値を設定します。詳細については、「ノードのスケジューリングプロパティ」をご参照ください。

   説明 DataWorks では、ノード出力を使用して、ノードとその子ノード間のスケジューリング依存関係を設定します。SQL ノードが同期ノードの出力テーブルを処理する場合、自動解析機能により、テーブルリネージに基づいて同期ノードが SQL ノードの親ノードとして設定されます。「ノード出力」で指定された <MaxCompute プロジェクト名（本番環境）>.ods_user_info_d_odps が存在することを確認してください。

   プロパティ	値	スクリーンショット
   スケジューリングパラメーター	$bizdate（パラメーター名：bizdate；前日の日付を yyyymmdd 形式で照会）
   スケジューリング周期	日
   スケジュール時刻	00:30
   再実行	実行状態に関係なく許可
   リソースグループ	ご利用のサーバーレスリソースグループ
   依存関係	workshop_start が [親ノード] に表示されていることを確認します。表示されていない場合は、ステップ 1 のワークフロー設計を再確認してください。また、<本番環境の MaxCompute プロジェクト名>.ods_user_info_d_odps という名前のノード出力が存在することを確認します。存在しない場合は、手動で追加してください。

OSS ログの MaxCompute への同期

このタスクは、HttpFile ソースから user_log.txt ファイルを読み取り、MaxCompute テーブル ods_raw_log_d_odps へ書き込みます。

ログファイルは、以下の形式で各行に 1 件のアクセスレコードを格納します。

$remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent"$http_referer" "$http_user_agent" [unknown_content];

1. ods_raw_log_d_odps ノードをダブルクリックして、構成タブを開きます。

2. ソース、リソースグループ、送信先を構成し、次へ をクリックして接続性テストを完了します。

   パラメーター	値
   ソース	HttpFile — データソース名：user_behavior_analysis_HttpFile
   リソースグループ	ご利用のサーバーレスリソースグループ
   送信先	MaxCompute — データソース名：user_behavior_analysis_mysql

   

3. ソースおよび送信先の設定を構成します。ソースを構成した後、データ構造の確認 をクリックして、ログファイルが読み取れることを確認します。デフォルトのフィールドマッピングおよびチャネル制御設定を使用します。利用可能なすべてのオプションについては、「コードレス UI を使用したバッチ同期タスクの構成」をご参照ください。

   項目	パラメーター	値	スクリーンショット
   ソース	ファイルパス	/user_log.txt
   	ファイルタイプ	text
   	列区切り文字	|
   高度な構成	コーディング	UTF-8
   	圧縮形式	UTF-8
   	ヘッダーをスキップ	いいえ
   送信先	Tunnel リソースグループ	共通伝送リソース（デフォルト）
   	スキーマ	default
   	テーブル	ods_raw_log_d_odps
   	パーティション情報	${bizdate}
   	書き込みモード	上書き挿入（書き込み前に既存のデータをクリア）
   	空文字列を NULL に変換して書き込み	いいえ

4. スケジューリングプロパティを構成します。右側ナビゲーションペインで プロパティ をクリックし、以下の値を設定します。詳細については、「ノードのスケジューリングプロパティ」をご参照ください。

   説明 ノード出力出力名 <本番環境の MaxCompute プロジェクト名>.ods_raw_log_d_odps が存在することを確認してください。DataWorks は、この出力を下流の SQL ノードが同期ノードに依存する際に 自動解析 で使用します。

   属性	値	スクリーンショット
   スケジューリングパラメーター	$bizdate (パラメーター名: bizdate; 前日の日付を yyyymmdd 形式で照会します)
   スケジューリングサイクル	Day
   スケジュール時刻	00:30
   再実行	実行ステータスに関わらず許可
   リソースグループ	ご利用のサーバーレスリソースグループ
   依存関係	workshop_start が [親ノード] に表示されていることを確認します。また、<本番環境の MaxCompute プロジェクト名>.ods_raw_log_d_odps というノード出力が存在することを確認します。存在しない場合は手動で追加します。

ワークフローの実行

1. DataStudio で、ビジネスフロー 下の User profile analysis_MaxCompute ワークフローをダブルクリックします。

2. ワークフローキャンバスで、ツールバーの アイコンをクリックして、依存関係順にすべてのノードを実行します。

3. ステータスを確認します：

   • ノードのステータス： 状態のノードは、正常に実行されたことを示します。

   • ノードのログ： ods_user_info_d_odps または ods_raw_log_d_odps を右クリックし、ログの表示 を選択します。ログに以下のような出力が表示された場合、データの同期が完了しています。

結果の検証

正常に実行された後：

• 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 に格納されます。

データが正しく読み込まれたことを確認するために、アドホッククエリで行数を照会します。

1. DataStudio の左側ナビゲーションウィンドウで、 アイコンをクリックします。アドホッククエリ ペインで、アドホッククエリ を右クリックし、ノードの作成 > ODPS SQL を選択します。

2. 以下の SQL を実行して、行数を確認します。データタイムスタンプ を実行時のデータタイムスタンプ（例：ワークフローが 2023 年 6 月 21 日に実行された場合、データタイムスタンプは 20230620）に置き換えます。

   select count(*) from ods_user_info_d_odps where dt='データタイムスタンプ';
   select count(*) from ods_raw_log_d_odps where dt='データタイムスタンプ';