DataWorks:データの同期

MySQL データソースの追加 (user_behavior_analysis_mysql)

このチュートリアルでは、基本ユーザー情報は MySQL データベースに格納されています。基本ユーザー情報データ（ods_user_info_d）を StarRocks に同期するには、MySQL データソースを作成する必要があります。

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

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

   2. 左側のナビゲーションウィンドウで、Data Sourceをクリックします。

2. Add Connection をクリックし、MySQL を検索して選択します。

3. [MySQL データソースの追加] ページで、開発環境と本番環境の両方に、次のサンプル値を使用します。

   次の表に、主要なパラメーターを示します。その他のパラメーターにはデフォルト値を使用できます。

   パラメーター	説明
   Data Source Name	user_behavior_analysis_mysql を入力します。
   Description	DataWorks チュートリアル専用のデータソースです。これにより、単一テーブルのオフラインデータ同期タスクを設定する際に、提供されているテストデータにアクセスできます。このデータソースは読み取り専用で、Data Integration でのみ利用可能です。
   設定モード	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. 接続設定 セクションで、開発環境と本番環境の両方に対して テスト接続 をクリックします。接続ステータスが 接続可能 であることを確認します。

   重要

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

   • 利用可能なリソースグループがない場合は、Go to Buy セクションの画面の指示に従い、Go to Buyと購入したリソースグループのバインドをクリックします。

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

HttpFile データソースの追加 (user_behavior_analysis_httpfile)

このチュートリアルでは、ユーザーの Web サイトアクセスログデータは OSS に保存されています。OSS から StarRocks にユーザーの Web サイトアクセスログ (user_log.txt) を同期するために、Httpfile データソースを作成します。

1. Management Center ページで、左側のナビゲーションウィンドウの Data Source をクリックします。

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

3. [HttpFile データソースの追加] ページで、開発環境と本番環境の両方で以下のサンプル値を使用します。

   次の表に、主要なパラメーターを示します。その他のパラメーターにはデフォルト値を使用できます。

   パラメーター	説明
   Data Source Name	user_behavior_analysis_httpfile と入力します。
   Description	DataWorks チュートリアル専用のデータソースです。これにより、単一テーブルのオフラインデータ同期タスクを設定する際に、提供されているテストデータにアクセスできます。このデータソースは読み取り専用で、Data Integration でのみ利用可能です。
   URL	開発環境と本番環境の両方で、[URL] を https://dataworks-workshop-2024.oss-cn-shanghai.aliyuncs.com に設定します。

4. 接続設定 セクションで、開発環境と本番環境の両方で テスト接続 をクリックし、接続ステータスが 接続可能 であることを確認します。

   重要

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

   • 利用可能なリソースグループがない場合は、Go to Buy セクションの画面の指示に従い、Go to Buy と 購入したリソースグループのバインド をクリックします。

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

初期ノードの設定

1. フローオーケストレーションページで、workshop_start_starrocks ノードにマウスカーソルを合わせ、Open Node をクリックします。

2. workshop_start_starrocks ノード構成ページの右側で、Scheduling Configuration をクリックしてスケジューリングプロパティを構成します。 次の表では、このチュートリアルの主要なパラメーターについて説明します。 その他のパラメーターについては、デフォルト値のままにしておくことができます。

   パラメーター	説明
   Scheduling Type	このチュートリアルでは、このパラメーターを dry run に設定します。
   Scheduling Resource Groups	「環境の準備」ステップで作成したサーバーレスリソースグループを選択します。
   Node dependency configuration	workshop_start_starrocks は初期ノードであり、上流の依存関係がないため、Using the Workspace Root Node をクリックして、ワークスペースのルートノードからワークフローをトリガーします。 ワークスペースルートノードは、WorkspaceName_root というフォーマットで命名されます。

ユーザーテーブルを作成 ddl_ods_user_info_d_starrocks

StarRocks にデータを同期する前に、MySQL データソースのユーザー基本情報を格納する ddl_ods_user_info_d_starrocks テーブルを作成します。このテーブルは、後述するようにノード内で作成することも、[Data Catalog] で手動で作成することもできます。

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

2. CREATE TABLE 文を編集します。

   CREATE TABLE IF NOT EXISTS ods_user_info_d_starrocks (
       uid STRING COMMENT 'ユーザー ID',
       gender STRING COMMENT '性別',
       age_range STRING COMMENT '年齢層',
       zodiac STRING COMMENT '星座',
       dt STRING not null COMMENT '日付'
   ) 
   DUPLICATE KEY(uid)
   COMMENT 'ユーザー行動分析 - 基本ユーザー情報テーブル'
   PARTITION BY(dt) 
   PROPERTIES("replication_num" = "1");

3. デバッグ設定を構成します。

   StarRocks ノード構成ページの右側で、Run Configuration をクリックし、次のパラメーターを構成します。ステップ 4 では、Run Configuration のパラメーターを使用してデバッグ中に実行をテストできます。

   パラメーター	説明
   Computing Resources	「環境の準備」ステップでバインドした StarRocks コンピューティングリソースを選択します。
   Resource Group	「環境の準備」ステップで作成したサーバーレスリソースグループを選択します。

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

   ノード構成ページの右側で、Scheduling Configuration をクリックしてスケジューリングプロパティを表示します。このチュートリアルでは、デフォルト値を使用できます。これらのパラメーターの詳細については、「ノードスケジューリング構成」をご参照ください。

   • Scheduling Parameters: ワークフローレベルで構成するため、ノードレベルでの構成は不要です。タスクやコードで直接参照できます。

   • Scheduling Policy: Delayed execution time パラメーターを使用して、ワークフローの開始後、子ノードを実行するまでの遅延時間を指定できます。 このチュートリアルでは、この設定は使用しません。

5. 上部ツールバーで、Save をクリックします。

ログテーブルの作成: ddl_ods_raw_log_d_starrocks

StarRocks にデータを同期する前に、HttpFile データソースからのユーザーのウェブサイトアクセスログを保存するための ddl_ods_raw_log_d_starrocks テーブルを作成します。 テーブルは、後述するようにノード内で作成するか、[Data Catalog] で手動で作成できます。

1. フローオーケストレーションページで、ddl_ods_raw_log_d_starrocks ノードにカーソルを合わせ、Open Node をクリックします。

2. CREATE TABLE 文を編集します。

   CREATE TABLE IF NOT EXISTS ods_raw_log_d_starrocks (
       col STRING COMMENT 'ログエントリ',
       dt DATE  not null COMMENT '日付'
   ) DUPLICATE KEY(col) 
   COMMENT 'ユーザー行動分析 - 生のウェブサイトアクセスログテーブル' 
   PARTITION BY(dt) 
   PROPERTIES ("replication_num" = "1");

3. デバッグ設定を構成します。

   バッチ同期ノード構成ページの右側で、Run Configuration をクリックし、ステップ 4 のデバッグ実行中に Run Configuration の関連パラメーターでテスト実行を行うように、以下のパラメーターを構成します。

   パラメーター	説明
   Computing Resources	「環境の準備」ステップでバインドした StarRocks コンピューティングリソースを選択します。
   Resource Group	「環境の準備」ステップで作成したサーバーレスリソースグループを選択します。

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

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

   • Scheduling Parameters：ワークフローレベルで構成するため、ノードレベルの構成は不要です。タスクやコードで直接参照できます。

   • Scheduling Policy: Delayed execution time パラメーターを使用すると、ワークフローの開始後に子ノードの実行を遅延させることができます。このチュートリアルでは、この設定は使用しません。

5. 上部のツールバーで、Save をクリックします。

ユーザーデータパイプラインの設定 (ods_user_info_d_starrocks)

1. フローオーケストレーションページで、ods_user_info_d_starrocks ノードにマウスカーソルを合わせ、Open Node をクリックします。

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

   パラメーター	説明
   Data source	Data source： MySQL Data Source Name: user_behavior_analysis_mysql
   My Resource Group	「環境の準備」ステップで作成したサーバーレスリソースグループを選択します。
   Data going	Data going： StarRocks Data Source Name: doc_starrocks_storage_compute_tightly_01

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

   • Configure Source and Destination。

     セクション	パラメーター	設定
     Data source	Table	MySQL テーブル ods_user_info_d を選択します。
     	Shard Key	プライマリキーまたはインデックス付きの列を分割キーとして使用します。整数データ型のみがサポートされています。 このチュートリアルでは、スプリットキーを uid フィールドに設定します。
     Data going	Table	StarRocks テーブル ods_user_info_d_starrocks を選択します。
     	Statement Run Before Writing	このチュートリアルでは、dt フィールドに基づく動的パーティションを使用します。ノードが再実行された際のデータの重複を防ぐため、次の SQL ステートメントは各同期の前にターゲットパーティションを削除します。 ALTER TABLE ods_user_info_d_starrocks DROP PARTITION IF EXISTS p${var} FORCE 文では、スケジューリング構成時に値が割り当てられるパラメーター ${var} が使用されます。 これにより、スケジューリングシナリオでの動的パラメーター入力が可能になります。 詳細については、「スケジューリング設定」をご参照ください。
     	Streamload Request Parameters	StreamLoad のリクエストパラメーター。値は JSON 形式である必要があります。 { "row_delimiter": "\\x02", "column_separator": "\\x01" }

   • Field Mapping。

     フィールドマッピングを使用して、ソースフィールドと送信先フィールドの関係を定義します。スケジューリングパラメーターを T 変数に割り当てることで、StarRocks パーティションフィールドを動的に入力し、日次データが正しいパーティションに書き込まれるようにします。

     1. [同名のフィールドをマッピング] をクリックします。システムは、ソースの MySQL フィールドを同じ名前の送信先フィールドに自動的にマッピングします。

     2. [追加] をクリックし、'${var}' を入力し、このフィールドを StarRocks テーブルの dt フィールドに手動でマップします。

   • Channel。

     このチュートリアルでは、Policy for Dirty Data Records を Disallow Dirty Data Records に設定し、その他の設定はデフォルト値のままにします。 詳細については、「コードレス UI でバッチ同期ノードを設定する」をご参照ください。

4. デバッグ設定を構成します。

   バッチ同期ノード設定ページの右側で、Run Configuration をクリックし、次のパラメーターを設定します。ステップ 4 のデバッグ実行中に、Run Configuration セクションのパラメーターを使用して実行をテストできます。

   パラメーター	説明
   Resource Group	「環境の準備」ステップで作成したサーバーレスリソースグループを選択します。
   Script Parameters	Add parameter をクリックし、var=20250223 のように、定数を var=yyyymmdd フォーマットで設定します。 デバッグ中、 Data Studio はタスク変数をこの定数で置き換えます。

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

   ノード構成ページの右側で、Scheduling Configuration をクリックすると、スケジューリングプロパティが表示されます。このチュートリアルでは、デフォルト値のままにしておくことができます。これらのパラメーターの詳細については、「ノードスケジューリング構成」をご参照ください。

   • Scheduling Parameters: これらはワークフローレベルで構成するため、ノードレベルの構成は不要です。タスクまたはコードで直接参照できます。

   • Scheduling Policy: Delayed execution time パラメーターを使用して、ワークフローの開始後に子ノードを実行するための遅延を指定できます。このチュートリアルでは、この設定は使用しません。

6. 上部ツールバーで、Save をクリックします。

ログデータパイプラインの設定 (ods_raw_log_d_starrocks)

1. ワークフローオーケストレーションページで、ods_raw_log_d_starrocks ノードにポインターを合わせ、Open Node をクリックします。

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

   以下の詳細に従ってData source、My Resource Group、Data goingを設定します。その後、次のステップをクリックし、画面の指示に従って接続性テストを完了します。

   パラメーター	設定
   Data source	Data source： HttpFile Data Source Name: user_behavior_analysis_HttpFile
   My Resource Group	「環境の準備」ステップで作成したサーバーレスリソースグループを選択します。
   Data going	Data going: StarRocks Data Source Name： doc_starrocks_storage_compute_tightly_01

3. 次のステップ をクリックして、同期タスクを設定します。

   • Configure Source and Destination。

     セクション	パラメーター	設定
     Data source	File Path	/user_log.txt
     	File Type	テキスト
     	Field Delimiter	|
     	Advanced Settings > Skip Header	No
     	ソースを設定した後、Confirm Data Structure をクリックします。
     Data going	Table	StarRocks テーブル ods_raw_log_d_starrocks を選択します。
     	Statement Run Before Writing	このチュートリアルでは、dt フィールドに基づく動的パーティションを使用します。ノードが再実行された際のデータ重複を防ぐため、次の SQL ステートメントは、各同期の前にターゲットパーティションを削除します。 ALTER TABLE ods_raw_log_d_starrocks DROP PARTITION IF EXISTS p${var} FORCE 文は、パラメーター ${var} を使用します。このパラメーターには、スケジューリング構成時に値が割り当てられます。これにより、スケジューリングシナリオでの動的パラメータ入力が可能になります。詳細については、「スケジューリング設定」をご参照ください。
     	Streamload Request Parameters	StreamLoad のリクエストパラメーター。値は JSON 形式である必要があります。 { "row_delimiter": "\\x02", "column_separator": "\\x01" }

   • Channel。

     このチュートリアルでは、Policy for Dirty Data Records を Disallow Dirty Data Records に設定し、その他の設定はデフォルト値のままにします。詳細については、「コードレス UI でバッチ同期ノードを設定する」をご参照ください。

   • Field Mapping。

     ノードのツールバーで、 アイコンをクリックして、構成モードを Wizard Mode から Script Mode に切り替えます。これにより、HttpFile データソースのフィールドをマップし、StarRocks パーティションフィールド dt に値を動的に割り当てることができます。

     • ソースの column 設定に、次のコードを追加します：

        {
                  "type": "STRING",
                 "value": "${var}"
               }

     • 以下は、ods_raw_log_d_starrocks ノードの完全なスクリプトです：

       {
           "type": "job",
           "version": "2.0",
           "steps": [
               {
                   "stepType": "httpfile",
                   "parameter": {
                       "fileName": "/user_log.txt",
                       "nullFormat": "",
                       "compress": "",
                       "requestMethod": "GET",
                       "connectTimeoutSeconds": 60,
                       "column": [
                           {
                               "index": 0,
                               "type": "STRING"
                           },
                           {
                               "type": "STRING",
                               "value": "${var}"
                           }
                       ],
                       "skipHeader": "false",
                       "encoding": "UTF-8",
                       "fieldDelimiter": "|",
                       "fieldDelimiterOrigin": "|",
                       "socketTimeoutSeconds": 3600,
                       "envType": 0,
                       "datasource": "user_behavior_analysis",
                       "bufferByteSizeInKB": 1024,
                       "fileFormat": "text"
                   },
                   "name": "Reader",
                   "category": "reader"
               },
               {
                   "stepType": "starrocks",
                   "parameter": {
                       "loadProps": {
                           "row_delimiter": "\\x02",
                           "column_separator": "\\x01"
                       },
                       "envType": 0,
                       "datasource": "doc_starrocks_storage_compute_tightly_01",
                       "column": [
                           "col",
                           "dt"
                       ],
                       "tableComment": "",
                       "table": "ods_raw_log_d_starrocks",
                       "preSql": "ALTER TABLE ods_raw_log_d_starrocks DROP PARTITION IF EXISTS  p${var} FORCE;"
                   },
                   "name": "Writer",
                   "category": "writer"
               },
               {
                   "copies": 1,
                   "parameter": {
                       "nodes": [],
                       "edges": [],
                       "groups": [],
                       "version": "2.0"
                   },
                   "name": "Processor",
                   "category": "processor"
               }
           ],
           "setting": {
               "errorLimit": {
                   "record": "0"
               },
               "locale": "en",
               "speed": {
                   "throttle": false,
                   "concurrent": 2
               }
           },
           "order": {
               "hops": [
                   {
                       "from": "Reader",
                       "to": "Writer"
                   }
               ]
           }
       }

4. デバッグ設定を構成します。

   バッチ同期ノードの構成ページの右側で、Run Configuration をクリックし、以下のパラメーターを設定します。ステップ 4 のデバッグ実行中に、Run Configuration セクションのパラメーターを使用して実行をテストできます。

   パラメーター	説明
   Resource Group	「環境の準備」ステップで作成したサーバーレスリソースグループを選択します。
   Script Parameters	Add parameter をクリックし、定数を var=yyyymmdd フォーマットで設定します (例: var=20250223)。 デバッグ中に、DataStudio はタスク変数をこの定数に置き換えます。

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

   ノード構成ページの右側で、Scheduling Configuration をクリックしてスケジューリングプロパティを表示します。このチュートリアルでは、デフォルト値のままにしておくことができます。これらのパラメーターの詳細については、「ノードのスケジューリング構成」をご参照ください。

   • Scheduling Parameters: これらはワークフローレベルで構成するため、ノードレベルの構成は必要ありません。 タスクまたはコードで直接参照できます。

   • Scheduling Policy: Delayed execution time パラメーターを使用して、ワークフローの開始後に子ノードの実行に遅延を指定できます。このチュートリアルでは、この設定は使用しません。

6. 上部のツールバーで、Save をクリックします。