DataWorks:ワークフローでのデータプッシュノードの設定に関するベストプラクティス

ステップ 1: データプッシュフローを設定する

DataPushDemo ワークフローをダブルクリックします。ワークフローキャンバスで、SalesAmountPreMonth、Condition、CompliantData、Top3Categories、NonCompliantData、Bottom3Categories ノードを接続して、データプッシュフローを形成します。次の図は、データプッシュフローを示しています。

ステップ 2: SQL クエリノードを設定する

SQL クエリノードを設定してテストデータをクエリし、[入力パラメーターと出力パラメーター] セクションに outputs パラメーターを追加して、クエリ結果を分岐ノードに渡すことができます。

1. SalesAmountPreMonth ノードをダブルクリックします。ノードの設定タブで、ノードのコードを記述します。サンプルコード:

   -- 前月の総売上高をクエリします。
   SELECT SUM(sales) AS sales_amount
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59');

2. ノードの構成タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを構成します。

   • スケジュール時刻: 値を 08:00 に設定します。

   • リソースグループ: 既存のサーバーレスリソースグループを選択します。

   • 親ノード: [ルートノードを追加] を選択します。

   • 出力パラメーター: [入力パラメーターと出力パラメーター] セクションで、[出力パラメーター] の右側にある [代入パラメーターを追加] をクリックして、SQL クエリノードの outputs パラメーターを追加します。 outputs パラメーターは、データプッシュノードの入力パラメーターとして使用されます。

ステップ 3: 分岐ノードを設定する

[入力パラメーターと出力パラメーター] セクションで、SQL クエリノードの outputs パラメーターを分岐ノードの 入力パラメーターとして設定できます。分岐ノードは論理判断を実行した後、outputs パラメーターを使用して、クエリ結果を分岐ノードの子ノードに渡します。

1. Condition ノードをダブルクリックします。ノードの構成タブの定義セクションで、[ブランチの追加] をクリックします。[ブランチ定義] ダイアログボックスで、パラメーターを構成します。次の表は、構成する必要があるパラメーターについて説明しています。

   パラメーター	適合データを受け取る子ノード	不適合データを受け取る子ノード
   条件	${inputs[0][0]}>=500000	${inputs[0][0]}<500000
   関連付けられたノード出力	適合	不適合
   説明	特定の条件を満たす売上高	特定の条件を満たさない売上高

   説明

   [0][0] は、分岐ノードの先祖ノードの outputs パラメーターのデータを指定する 2 次元配列です。

   • 先祖ノードが SQL クエリノードの場合、2 次元配列を使用して、分岐ノードの先祖ノードの outputs パラメーターのデータを指定します。

   • 先祖ノードが Python ノードの場合、1 次元配列を使用して、分岐ノードの先祖ノードの outputs パラメーターのデータを指定します。

2. ブランチノードの構成タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを構成します。次の表は、構成する必要があるパラメーターについて説明しています。

   セクションとパラメーター		説明
   スケジュール	スケジューリングサイクル	値を 日 に設定します。
   	スケジュール時刻	値を 08:00 に設定します。
   	再実行	値を 実行状態に関係なく許可 に設定します。
   リソースグループ	リソースグループ	スケジューリング用に既存のリソースグループを選択します。 説明 データプッシュノードを初めて使用するときは、チケットを送信 して、スケジューリング用にリソースグループをアップグレードする必要があります。
   現在のノードの出力名		分岐を設定すると、出力名が自動的に解析されて表示されます。出力名 は、分岐を追加したときに [関連付けられたノード出力] パラメーターに設定した値と同じです。
   入力パラメーターと出力パラメーター	入力パラメーター	[作成] をクリックします。パラメーター名: inputs。 値のソース: SalesAmountPreMonth ノードの outputs パラメーターを選択します。
   	出力パラメーター	デフォルトでは、outputs パラメーターが追加されます。

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 4: 分岐ノードの子ノードを設定する

この例では、CompliantData ノードと NonCompliantData ノードは分岐ノードの子ノードです。[分岐ノード] が outputs パラメーターを使用してクエリ結果を子ノードに渡した後、各子ノードの [入力パラメーターと出力パラメーター] セクションに outputs パラメーターを追加して、クエリ結果を対応するデータプッシュノードに渡すことができます。

1. CompliantData ノードと NonCompliantData ノードをそれぞれダブルクリックします。各ノードの設定タブで、ノードのコードを記述します。

   CompliantData ノードのサンプルコード

   SET @all_cat_sales_volume_month := 0.0;
   SELECT SUM(sales) INTO @all_cat_sales_volume_month FROM orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59');
   
   -- 一時テーブルを作成します。
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array (
     category VARCHAR(255),
     sales DOUBLE,
     all_cat_sales_volume_month DOUBLE
   );
   -- データをクエリし、一時テーブルに書き込みます。
   INSERT INTO temp_array (category, sales, all_cat_sales_volume_month) SELECT category, SUM(sales) AS amount, @all_cat_sales_volume_month FROM orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59') GROUP BY category ORDER BY amount DESC limit 3;
   -- 売上高が特定の条件を満たす上位 3 つのカテゴリをクエリします。
   SELECT category, sales, all_cat_sales_volume_month FROM temp_array;

   NonCompliantData ノードのサンプルコード

   SET @all_cat_sales_volume_month := 0.0;
   SELECT SUM(sales) INTO @all_cat_sales_volume_month FROM orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59');
   
   -- 一時テーブルを作成します。
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array (
     category VARCHAR(255),
     sales DOUBLE,
     all_cat_sales_volume_month DOUBLE
   );
   
   -- データをクエリし、一時テーブルに書き込みます。
   INSERT INTO temp_array (category, sales, all_cat_sales_volume_month) SELECT category, SUM(sales) AS amount, @all_cat_sales_volume_month FROM orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59') GROUP BY category ORDER BY amount ASC limit 3;
   
   -- 売上高が特定の条件を満たさない下位 3 つのカテゴリをクエリします。
   SELECT category, sales, all_cat_sales_volume_month FROM temp_array;

2. ノードの構成タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを構成します。

   • スケジュール時刻: 値を 08:00 に設定します。

   • リソースグループ: 既存のサーバーレスリソースグループを選択します。

   • 親ノード: 分岐ノードとその子ノード間の依存関係が正しいかどうかを確認します。依存関係は、ステップ 1 でデータプッシュフローを設定するときに設定されます。

     • CompliantData: [依存関係] セクションで、[上位ノードの出力名] 列の値が [準拠] であることを確認します。

     • NonCompliantData: [依存関係] セクションで、[上位ノードの出力名] 列の値が [非準拠] であるかどうかを確認します。

   • 出力パラメーター: [入力パラメーターと出力パラメーター] セクションで、[出力パラメーター] の右側にある [代入パラメーターを追加] をクリックして、SQL クエリノードの outputs パラメーターを追加します。 outputs パラメーターは、データプッシュノードの入力パラメーターとして使用されます。

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 5: データプッシュノードを設定する

各データプッシュノードの [入力パラメーターと出力パラメーター] セクションの 入力パラメーター に入力パラメーターを追加して、CompliantData ノードと NonCompliantData ノードの outputs パラメーターを取得できます。このようにして、データプッシュノードは入力パラメーターを使用してデータを特定の宛先にプッシュできます。

1. Top3Categories ノードと Bottom3Categories ノードをそれぞれダブルクリックします。関連ノードの設定タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを設定します。次の表に、設定する必要があるパラメーターを示します。

   セクションとパラメーター		説明	スクリーンショット
   スケジューリングパラメーター	パラメーター名	値を curdate に設定します。
   	パラメーター値	値を $[yyyymmddhh:mi:ss] に設定します。
   スケジュール	スケジューリングサイクル	値を 日 に設定します。
   	スケジュール時刻	値を 08:00 に設定します。 説明 この例では、毎日 08:00 に特定の宛先にデータがプッシュされるようにします。このパラメーターは、ビジネス要件に基づいて設定できます。
   	再実行	値を 実行状態に関係なく許可 に設定します。
   リソースグループ	リソースグループ	スケジューリング用に既存のリソースグループを選択します。 説明 データプッシュノードを初めて使用するときは、チケットを送信 して、スケジューリング用にリソースグループをアップグレードする必要があります。
   入力パラメーターと出力パラメーター	入力パラメーター	[作成] をクリックして、入力パラメーターを追加します。 パラメーター名: inputs。 値のソース: Top3Categories: ドロップダウンリストから CompliantData ノードの outputs パラメーターを選択します。 Bottom3Categories: ドロップダウンリストから NonCompliantData ノードの outputs パラメーターを選択します。	Top3Categories Bottom3Categories

2. プッシュするデータを設定します。

   • 宛先: [宛先] ドロップダウンリストから宛先を選択します。使用可能な宛先がない場合は、[宛先の作成] をクリックして宛先を作成します。次の表に、宛先を作成するためのパラメーターを示します。

     パラメーター	説明
     タイプ	プッシュチャネルを選択します。DingTalk、Lark、WeCom、Microsoft Teams、電子メールがサポートされています。
     名前	ビジネス要件に基づいて名前を入力します。
     Webhook	選択したプッシュチャネルの Webhook URL を入力します。対応するプラットフォームで関連するプッシュチャネルの URL を取得する必要があります。 説明 Lark の Webhook URL の取得方法については、「Webhook トリガーを使用する」をご参照ください。 Teams の Webhook URL の取得方法については、「Microsoft Teams 用のワークフローで受信 Webhook を作成する」をご参照ください。

   • タイトル: Top3Categories ノードには 売上高が上位 3 位のカテゴリ、Bottom3Categories ノードには 売上高が下位 3 位のカテゴリ と入力します。

   • 本文: ビジネス要件に基づいてコンテンツを設定します。詳細については、「データプッシュ」トピックの「プッシュするコンテンツを設定する」セクションをご参照ください。

     説明

     [本文] セクションでは、SQL クエリノードによってクエリされたフィールド名をプレースホルダーとして直接使用して、データプッシュノードの入力パラメーターを取得できます。

     • Top3Categories ノードのサンプル設定

     • Bottom3Categories ノードのサンプル設定

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 6: データプッシュフローをテストする

条件付きデータプッシュフローを設定したら、ワークフローをコミットしてワークフロー内のノードをデプロイする前に、フローをテストする必要があります。

1. DataPushDemo ワークフローをダブルクリックします。

2. ワークフローの設定タブで、SalesAmountPreMonth ノードを右クリックし、[現在のノードとその子孫ノードを実行] を選択します。

   説明

   データプッシュフロー内のノードでエラーが発生した場合は、ノードを右クリックして [ログの表示] を選択し、ログを表示します。

   

ステップ 1: データプッシュフローを設定する

DataPushDemo ワークフローをダブルクリックします。ワークフローキャンバスで、SalesAmountPreWeek、Top3CategoryList、Top3CategoriesPreWeek ノードを接続して、データプッシュフローを形成します。次の図は、データプッシュフローを示しています。

ステップ 2: SQL クエリノードを設定する

SQL クエリノードを設定してテストデータをクエリし、[入力パラメーターと出力パラメーター] セクションに outputs パラメーターを追加して、SQL クエリノードのクエリ結果を代入ノードに渡すことができます。

1. SalesAmountPreWeek ノードをダブルクリックします。ノードの設定タブで、ノードのコードを記述します。サンプルコード:

   -- 前週の総売上高をクエリします。
   SELECT category, SUM(sales) AS amount
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 WEEK), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 23:59:59')
   GROUP BY category ORDER BY amount DESC limit 3;

2. ノードの構成タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを構成します。

   • スケジュール時刻: 値を 08:00 に設定します。

   • リソースグループ: 既存のサーバーレスリソースグループを選択します。

   • 親ノード: [ルートノードを追加] を選択します。

   • 出力パラメーター: [入力パラメーターと出力パラメーター] セクションで、[出力パラメーター] の右側にある [代入パラメーターを追加] をクリックして、SQL クエリノードの outputs パラメーターを追加します。 outputs パラメーターは、データプッシュノードの入力パラメーターとして使用されます。

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 3: 代入ノードを設定する

[入力パラメーターと出力パラメーター] セクションで、SQL クエリノードの outputs パラメーターを代入ノードの 入力パラメーターとして設定できます。次に、入力パラメーターに値を再割り当てして、代入ノードの outputs パラメーターを生成します。代入ノードは outputs パラメーターを使用して、データプッシュノードにデータを渡します。

1. Top3CategoryList ノードをダブルクリックします。ノードの設定タブで、言語 に Python を選択し、ノードのコードを記述します。

   def main():
   
       From datetime import date
       today = date.today()
       formatted_date = today.strftime('%Y-%m-%d')
       
       msg = '統計日: ' + formatted_date + ' \\n\\n ' \
       '- 1: ${inputs[0][0]}, 売上高: ${inputs[0][1]} \\n\\n ' \
       '- 2: ${inputs[1][0]}, 売上高: ${inputs[1][1]} \\n\\n ' \
       '- 3: ${inputs[2][0]}, 売上高: ${inputs[2][1]} \\n\\n '
       
       print(msg)
   
   
   if __name__ == "__main__":
       import sys
       main()

2. ノードの構成タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを構成します。

   • スケジュール時刻: 値を 08:00 に設定します。

   • リソースグループ: スケジューリング用に既存のリソースグループを選択します。

   • 入力パラメーターと出力パラメーター

     • 入力パラメーター

       • [作成] をクリックします。パラメーター名: inputs。

       • 値のソース: ドロップダウンリストから SalesAmountPreWeek ノードの outputs パラメーターを選択します。

     • 出力パラメーター: デフォルトでは、outputs パラメーターが追加されます。

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 4: データプッシュノードを設定する

[入力パラメーターと出力パラメーター] セクションで、代入ノードの outputs パラメーターをデータプッシュノードの 入力パラメーターとして設定できます。次に、データプッシュノードは入力パラメーターを使用してデータを特定の宛先にプッシュします。

1. Top3CategoriesPreWeek ノードをダブルクリックします。ノードの設定タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを設定します。次の表に、設定する必要があるパラメーターを示します。

   セクションとパラメーター		説明	スクリーンショット
   スケジューリングパラメーター	パラメーター名	値を curdate に設定します。
   	パラメーター値	値を $[yyyymmddhh:mi:ss] に設定します。
   スケジュール	スケジューリングサイクル	値を 日 に設定します。
   	スケジュール時刻	値を 08:00 に設定します。 説明 この例では、毎日 08:00 に特定の宛先にデータがプッシュされるようにします。このパラメーターは、ビジネス要件に基づいて設定できます。
   	再実行	値を 実行状態に関係なく許可 に設定します。
   リソースグループ	リソースグループ	スケジューリング用に既存のリソースグループを選択します。 説明 データプッシュノードを初めて使用するときは、チケットを送信 して、スケジューリング用にリソースグループをアップグレードする必要があります。
   入力パラメーターと出力パラメーター	入力パラメーター	[作成] をクリックして、入力パラメーターを追加します。 パラメーター名: inputs。 値のソース: ドロップダウンリストから Top3CategoryList ノードの outputs パラメーターを選択します。

2. プッシュするデータを設定します。

   • 宛先: [宛先] ドロップダウンリストから宛先を選択します。使用可能な宛先がない場合は、[宛先の作成] をクリックして宛先を作成します。次の表に、宛先を作成するためのパラメーターを示します。

     パラメーター	説明
     タイプ	プッシュチャネルを選択します。DingTalk、Lark、WeCom、Teams がサポートされています。
     名前	ビジネス要件に基づいて名前を入力します。
     Webhook	選択したプッシュチャネルの Webhook URL を入力します。対応するプラットフォームで関連するプッシュチャネルの URL を取得する必要があります。 説明 Lark の Webhook URL の取得方法については、「Webhook トリガーを使用する」をご参照ください。 Teams の Webhook URL の取得方法については、「Microsoft Teams 用のワークフローで受信 Webhook を作成する」をご参照ください。

   • タイトル: 前週の売上高が上位 3 位のカテゴリ と入力します。

   • 本文: ビジネス要件に基づいてコンテンツを設定します。詳細については、「データプッシュ」トピックの「プッシュするコンテンツを設定する」セクションをご参照ください。

     説明

     本文 セクションでは、SQL クエリノードによってクエリされたフィールド名をプレースホルダーとして直接使用して、データプッシュノードの入力パラメーターを取得できます。

     

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 5: データプッシュフローをテストする

スクリプトデータプッシュフローを設定したら、ワークフローをコミットしてワークフロー内のノードをデプロイする前に、フローをテストする必要があります。

1. DataPushDemo ワークフローをダブルクリックします。

2. ワークフローの設定タブで、SalesAmountPreWeek ノードを右クリックし、[現在のノードとその子孫ノードを実行] を選択します。

   説明

   データプッシュフロー内のノードでエラーが発生した場合は、ノードを右クリックして [ログの表示] を選択し、ログを表示します。

   

ステップ 1: データプッシュフローを設定する

DataPushDemo ワークフローをダブルクリックします。ワークフローキャンバスで、SalesAmountPreDay ノードと PushSalesAmountPreDay ノードを接続して、データプッシュフローを形成します。次の図は、データプッシュフローを示しています。

ステップ 2: SQL クエリノードを設定する

SQL クエリノードを設定してテストデータをクエリし、[入力パラメーターと出力パラメーター] セクションに outputs パラメーターを追加して、SQL クエリノードのクエリ結果をデータプッシュノードに渡すことができます。

1. SalesAmountPreDay ノードをダブルクリックします。ノードの設定タブで、ノードのコードを記述します。サンプルコード:

   -- temp_array という名前の一時テーブルを作成します。
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array (
     total_amount DOUBLE
   );
   
   -- 昨日の総売上高に関連するデータを temp_array テーブルに書き込みます。
   INSERT INTO temp_array (total_amount) 
   SELECT SUM(sales)
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 23:59:59');
   
   -- temp_array テーブルをクエリします。
   select total_amount FROM temp_array;

2. ノードの構成タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを構成します。

   • スケジュール時刻: 値を 08:00 に設定します。

   • リソースグループ: 既存のサーバーレスリソースグループを選択します。

   • 親ノード: [ルートノードを追加] を選択します。

   • 出力パラメーター: [入力パラメーターと出力パラメーター] セクションで、[出力パラメーター] の右側にある [代入パラメーターを追加] をクリックして、SQL クエリノードの outputs パラメーターを追加します。 outputs パラメーターは、データプッシュノードの入力パラメーターとして使用されます。

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 3: データプッシュノードを設定する

[入力パラメーターと出力パラメーター] セクションで、SQL クエリノードの outputs パラメーターをデータプッシュノードの 入力パラメーターとして設定できます。次に、データプッシュノードは入力パラメーターを使用してデータを特定の宛先にプッシュします。

1. PushSalesAmountPreDay ノードをダブルクリックします。ノードの設定タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを設定します。次の表に、設定する必要があるパラメーターを示します。

   セクションとパラメーター		説明	スクリーンショット
   スケジューリングパラメーター	パラメーター名	値を curdate に設定します。
   	パラメーター値	値を $[yyyymmddhh:mi:ss] に設定します。
   スケジュール	スケジューリングサイクル	値を 日 に設定します。
   	スケジュール時刻	値を 08:00 に設定します。 説明 この例では、毎日 08:00 に特定の宛先にデータがプッシュされるようにします。このパラメーターは、ビジネス要件に基づいて設定できます。
   	再実行	値を 実行状態に関係なく許可 に設定します。
   リソースグループ	リソースグループ	スケジューリング用に既存のリソースグループを選択します。 説明 データプッシュノードを初めて使用するときは、チケットを送信 して、スケジューリング用にリソースグループをアップグレードする必要があります。
   入力パラメーターと出力パラメーター	入力パラメーター	[作成] をクリックして、入力パラメーターを追加します。 パラメーター名: inputs。 値のソース: ドロップダウンリストから SalesAmountPreDay ノードの outputs パラメーターを選択します。

2. プッシュするデータを設定します。

   • 宛先: [宛先] ドロップダウンリストから宛先を選択します。使用可能な宛先がない場合は、[宛先の作成] をクリックして宛先を作成します。次の表に、宛先を作成するためのパラメーターを示します。

     パラメーター	説明
     タイプ	プッシュチャネルを選択します。DingTalk、Lark、WeCom、Teams がサポートされています。
     名前	ビジネス要件に基づいて名前を入力します。
     Webhook	選択したプッシュチャネルの Webhook URL を入力します。対応するプラットフォームで関連するプッシュチャネルの URL を取得する必要があります。 説明 Lark の Webhook URL の取得方法については、「Webhook トリガーを使用する」をご参照ください。 Teams の Webhook URL の取得方法については、「Microsoft Teams 用のワークフローで受信 Webhook を作成する」をご参照ください。

   • タイトル: 昨日の総売上高 と入力します。

   • 本文: ビジネス要件に基づいてコンテンツを設定します。詳細については、「データプッシュ」トピックの「プッシュするコンテンツを設定する」セクションをご参照ください。

     説明

     本文 セクションでは、SQL クエリノードによってクエリされたフィールド名をプレースホルダーとして直接使用して、データプッシュノードの入力パラメーターを取得できます。

     

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 4: フローをテストする

シンプルなデータプッシュフローを設定したら、ワークフローをコミットしてワークフロー内のノードをデプロイする前に、フローをテストする必要があります。

1. DataPushDemo ワークフローをダブルクリックします。

2. ワークフローの設定タブで、SalesAmountPreDay ノードを右クリックし、[現在のノードとその子孫ノードを実行] を選択します。

   説明

   データプッシュフロー内のノードでエラーが発生した場合は、ノードを右クリックして [ログの表示] を選択し、ログを表示します。

   

ステップ 1: データプッシュフローを設定する

DataPushDemo ワークフローをダブルクリックします。ワークフローキャンバスで、SalesAmountPreDay、SalesGrowthPreDay、CombinedPush ノードを接続して、データプッシュフローを形成します。次の図は、データプッシュフローを示しています。

ステップ 2: SQL クエリノードを設定する

各 SQL クエリノードを設定してテストデータをクエリし、[入力パラメーターと出力パラメーター] セクションの outputs パラメーターを使用して、SQL クエリノードのクエリ結果をデータプッシュノードに渡すことができます。

1. SalesGrowthPreDay ノードをクリックします。ノードの設定タブで、ノードのコードを記述します。サンプルコード:

   -- 一昨日分の売上合計金額を集計するための temp_array1 という名前のテーブルを作成します。
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array1 (
     category VARCHAR(255),
     sales DOUBLE
   );
   -- 一昨日分のデータを temp_array1 テーブルに書き込みます。
   INSERT INTO temp_array1 (category, sales) SELECT category, SUM(sales)
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 2 DAY), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 2 DAY), '%Y-%m-%d 23:59:59')
   GROUP BY category;
   
   -- 昨日分の売上合計金額を集計するための temp_array2 という名前のテーブルを作成します。
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array2 (
     category VARCHAR(255),
     sales DOUBLE
   );
   -- 昨日分のデータを temp_array2 テーブルに書き込みます。
   INSERT INTO temp_array2 (category, sales) SELECT category, SUM(sales)
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 23:59:59')
   GROUP BY category;
   
   -- 昨日分の売上増加金額を集計するための result という名前のテーブルを作成します。
   CREATE TEMPORARY TABLE IF NOT EXISTS result (
     category VARCHAR(255),
     diff DOUBLE
   );
   -- 売上増加データを result テーブルに書き込みます。
   INSERT INTO result (category, diff) SELECT temp_array2.category AS category, temp_array2.sales - temp_array1.sales AS diff FROM temp_array1 LEFT JOIN temp_array2 ON temp_array1.category = temp_array2.category;
   
   -- result テーブルからデータをクエリします。
   SELECT category, diff FROM result;

2. ノードの構成タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを構成します。

   • スケジュール時刻: 値を 08:00 に設定します。

   • リソースグループ: 既存のサーバーレスリソースグループを選択します。

   • 親ノード: [ルートノードを追加] を選択します。

   • 出力パラメーター: [入力パラメーターと出力パラメーター] セクションで、[出力パラメーター] の右側にある [代入パラメーターを追加] をクリックして、SQL クエリノードの outputs パラメーターを追加します。 outputs パラメーターは、データプッシュノードの入力パラメーターとして使用されます。

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 3: データプッシュノードを設定する

[入力パラメーターと出力パラメーター] セクションで、SalesAmountPreDay ノードと SalesGrowthPreDay ノードの outputs パラメーターをデータプッシュノードの 入力パラメーターとして設定できます。次に、データプッシュノードは入力パラメーターを使用してデータを特定の宛先にプッシュします。

1. CombinedPush ノードをダブルクリックします。ノードの設定タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを設定します。次の表に、設定する必要があるパラメーターを示します。

   セクションとパラメーター		説明	スクリーンショット
   スケジューリングパラメーター	パラメーター名	値を curdate に設定します。
   	パラメーター値	値を $[yyyymmddhh:mi:ss] に設定します。
   スケジュール	スケジューリングサイクル	値を 日 に設定します。
   	スケジュール時刻	値を 08:00 に設定します。 説明 この例では、毎日 08:00 に特定の宛先にデータがプッシュされるようにします。このパラメーターは、ビジネス要件に基づいて設定できます。
   	再実行	値を 実行状態に関係なく許可 に設定します。
   リソースグループ	リソースグループ	スケジューリング用に既存のリソースグループを選択します。 説明 データプッシュノードを初めて使用するときは、チケットを送信 して、スケジューリング用にリソースグループをアップグレードする必要があります。
   入力パラメーターと出力パラメーター	入力パラメーター	[作成] をクリックして、入力パラメーターを追加します。 パラメーター 1: パラメーター名: inputs1。 値のソース: ドロップダウンリストから SalesAmountPreDay ノードの outputs パラメーターを選択します。 パラメーター 2: パラメーター名: inputs2。 値のソース: ドロップダウンリストから SalesGrowthPreDay の outputs パラメーターを選択します。

2. プッシュするデータを設定します。

   • 宛先: [宛先] ドロップダウンリストから宛先を選択します。使用可能な宛先がない場合は、[宛先の作成] をクリックして宛先を作成します。次の表に、宛先を作成するためのパラメーターを示します。

     パラメーター	説明
     タイプ	プッシュチャネルを選択します。DingTalk、Lark、WeCom、Teams がサポートされています。
     名前	ビジネス要件に基づいて名前を入力します。
     Webhook	選択したプッシュチャネルの Webhook URL を入力します。対応するプラットフォームで関連するプッシュチャネルの URL を取得する必要があります。 説明 Lark の Webhook URL の取得方法については、「Webhook トリガーを使用する」をご参照ください。 Teams の Webhook URL の取得方法については、「Microsoft Teams 用のワークフローで受信 Webhook を作成する」をご参照ください。

   • タイトル: 昨日の売上高と売上高の増加 と入力します。

   • 本文: ビジネス要件に基づいてコンテンツを設定します。詳細については、「データプッシュ」トピックの「プッシュするコンテンツを設定する」セクションをご参照ください。

     説明

     本文 セクションでは、SQL クエリノードによってクエリされたフィールド名をプレースホルダーとして直接使用して、データプッシュノードの入力パラメーターを取得できます。

     

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 4: データプッシュフローをテストする

複合データプッシュフローとシンプルなデータプッシュフローを設定したら、ワークフローをコミットしてワークフロー内のノードをデプロイする前に、複合データプッシュフローをテストする必要があります。

1. DataPushDemo ワークフローをダブルクリックします。

2. ワークフローの設定タブで、CombinedPush ノードを右クリックし、[現在のノードとその子孫ノードを実行] を選択します。

   説明

   データプッシュフロー内のノードでエラーが発生した場合は、ノードを右クリックして [ログの表示] を選択し、ログを表示します。

   

ステップ 1: データプッシュフローを設定する

DataPushDemo ワークフローをダブルクリックします。ワークフローキャンバスで、MaxComputeDataSync、MaxComputeDataQuery、MaxComputeDataPush ノードを接続して、データプッシュフローを形成します。次の図は、データプッシュフローを示しています。

ステップ 2: バッチ同期ノードを設定する

MaxComputeDataSync ノードを構成して、データの準備 セクションで MySQL に書き込まれたテストデータを、将来使用するために作成された MaxCompute データソースに同期できます。

1. MaxComputeDataSync ノードをダブルクリックします。ノードの設定タブで、パラメーターを設定します。次の表にパラメーターを示します。

   項目		説明	スクリーンショット
   ソース	ソース	MySQL を選択します。
   	データソース名	追加した MySQL データソースを選択します。
   リソースグループ		サーバーレスリソースグループを選択します。
   宛先	宛先	MaxCompute(ODPS) を選択します。
   	データソース名	ワークスペースに追加されている MaxCompute データソースを選択します。

   設定が完了すると、システムはデータソースとリソースグループ間の接続性をテストします。データソースがネットワーク接続テストに合格したら、[次へ] をクリックして、ソースと宛先を設定します。

2. ソースと宛先を設定します。

   セクションとパラメーター		説明	スクリーンショット
   ソース	データソース	次の項目に注意してください。 デフォルト値の MySQL を保持します。 追加した MySQL データソースを選択します。
   	テーブル	orders テーブルを選択します。
   	データフィルタリング	このパラメーターは、ビジネス要件に基づいて設定します。この例では、このパラメーターは空のままです。
   	分割キー	ソーステーブルの列をシャードキーとして使用できます。プライマリキー列またはインデックス付き列を使用することをお勧めします。
   	データプレビュー	[データプレビュー] をクリックして、MySQL データソースから取得したデータが期待どおりかどうかを確認できます。
   宛先	データソース	次の項目に注意してください。 デフォルト値の MaxCompute を保持します。 ワークスペースに追加されている MaxCompute データソースを選択します。
   	トンネルリソースグループ	デフォルト値の 共有転送リソース を保持します。指定されたリソースは、トンネルクォータとして使用されます。
   	テーブル	[宛先テーブルスキーマを生成] をクリックして、宛先テーブルを生成します。
   	パーティション情報	日次増分データを対応する日付のパーティションに保存する場合は、日次増分データの同期用に pt パラメーターを設定できます。たとえば、pt を ${bizdate} に設定できます。
   	書き込みモード	[書き込み前に既存のデータをクリーンアップする (insert Overwrite)] を選択します。

3. ソースフィールドを、ソースフィールドと同じ名前の宛先フィールドにマッピングします。

4. チャネル制御ポリシーを設定します。

   • タスクの予想最大同時実行数: データ同期の際に使用される並列スレッドの実際の数は、リソースまたはタスクの特定の特性により、指定されたしきい値以下になる場合があります。使用する並列スレッドの実際の数に基づいて、選択したリソースグループに対して課金されます。この例では、このパラメーターを 2 に設定します。

   • [同期レート]: スロットリングを使用すると、ソースとデスティネーションの読み取りおよび書き込みワークロードを削減できます。スロットリングを有効にしない場合、システムは既存のハードウェア環境で使用可能な最大伝送パフォーマンスでノードを処理します。この例では、[電流制限なし] を選択します。

   • [不正なデータレコードのポリシー]: [不正なデータレコードを許可しない] を選択します。

   • 分散実行: デフォルトでは、このスイッチはオフになっています。指定した並列スレッドの最大数が 8 以上の場合にのみ、スイッチをオンにできます。

   

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

   ノードの設定タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを設定します。

   • スケジューリングパラメーター

     • パラメーター名: 値を bizdate に設定します。

     • パラメーター値: 値を $[yyyymmdd-1] に設定します。

   • スケジュール時刻: 値を 08:00 に設定します。

   • [再実行]: このパラメーターを「実行状態に関係なく許可」に設定します。

   • リソースグループ: スケジューリング用に既存のリソースグループを選択します。

   • [依存関係]: [ルートノードの追加] を選択して、ルートノードをバッチ同期ノードの先祖ノードとして使用します。

6. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 3: 代入ノードを設定する

MaxCompute データソースでは、ODPS SQL ノードを使用してデータをクエリし、[出力パラメーター] を使用してデータプッシュノードにデータを送信することはできません。 代入ノードを設定して MaxCompute データをクエリし、[入力と出力のパラメーター] セクションに [出力] パラメーターを追加して、クエリ結果をデータプッシュノードに渡すことができます。

1. MaxComputeDataQuery ノードをダブルクリックします。

   1. ノードの設定タブで、言語 ドロップダウンリストから ODPS SQL を選択します。

   2. ノードのコードを記述します。サンプルコード:

      -- 各パーティションの売上高を降順にランク付けするサブクエリを作成します。
      -- DENSE_RANK() 関数を呼び出して、パーティション内の各行にランクを割り当てます。パーティション内の複数の行の売上高が同じである場合、DENSE_RANK() 関数はそれらの行に同じランクを割り当てます。ランキング番号は連続しています。
      --
      -- サブクエリ:
      -- 1. orders テーブルから次の列を選択します。order_id、category、sales、datetime、pt。
      -- 2. DENSE_RANK() OVER (PARTITION BY pt ORDER BY sales DESC) 関数を呼び出します。
         -- PARTITION BY pt: pt フィールドに基づいてデータレコードをパーティション分割します。サブクエリは、pt フィールドで指定された各パーティションのデータレコードをランク付けします。
         -- ORDER BY sales DESC: 各パーティションのデータレコードを、sales フィールドの値に基づいて降順にランク付けします。
         -- rank 列には、各パーティションの各行の売上高のランクが格納されます。
      --
      -- メインクエリ:
      -- サブクエリのクエリ結果から、売上高が上位 3 位のデータレコードをクエリします。
      -- つまり、各データタイムスタンプの売上高が上位 3 位のデータレコードがクエリされます。
      
      SELECT
        order_id,          -- 注文 ID。
        category,          -- 製品カテゴリ。
        sales,             -- 売上高。
        datetime, -- 注文の支払い日時。
        pt -- データタイムスタンプ。
      FROM (
          SELECT
            order_id,
            category,
            sales,
            datetime,
            pt,
            DENSE_RANK() OVER (PARTITION BY pt ORDER BY sales DESC) AS rank  -- 各パーティションの売上高のランキングを計算します。
          FROM orders
          WHERE pt = '${bizdate}'  -- データタイムスタンプ (${bizdate} で指定) に基づいてデータレコードをクエリします。
      ) AS ranked_orders
      WHERE rank <= 3  -- 各パーティションの上位 3 つのデータレコードを保持します。

2. ノードの構成タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを構成します。

   • スケジュール時刻: 値を 08:00 に設定します。

   • リソースグループ: 既存のサーバーレスリソースグループを選択します。

   • [依存アップストリームノード]: Ancestor Node Name パラメーターの値が MaxComputeDataSync であるかどうかを確認します。

   • 出力パラメーター: 入力パラメーターと出力パラメーターのセクションで、出力パラメーターの右側にある [作成] をクリックして、代入ノードの出力パラメーターを追加します。出力パラメーターは、データプッシュノードの入力パラメーターとして使用されます。

     

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 4: データプッシュノードを設定する

[入力パラメーターと出力パラメーター] セクションで、代入ノードの outputs パラメーターをデータプッシュノードの 入力パラメーターとして設定できます。次に、データプッシュノードは入力パラメーターを使用してデータを特定の宛先にプッシュします。

1. MaxCompueDataPush ノードをダブルクリックします。ノードの設定タブの右側のナビゲーションウィンドウで、[プロパティ] タブをクリックします。[プロパティ] タブで、パラメーターを設定します。次の表に、設定する必要があるパラメーターを示します。

   セクションとパラメーター		説明	スクリーンショット
   スケジューリングパラメーター	パラメーター名	値を curdate に設定します。
   	パラメーター値	値を $[yyyymmddhh:mi:ss] に設定します。
   スケジュール	スケジューリングサイクル	値を 日 に設定します。
   	スケジュール時刻	値を 08:00 に設定します。 説明 この例では、毎日 08:00 に特定の宛先にデータがプッシュされるようにします。このパラメーターは、ビジネス要件に基づいて設定できます。
   	再実行	値を 実行状態に関係なく許可 に設定します。
   リソースグループ	リソースグループ	データプッシュ機能のリリース日である 2024 年 6 月 28 日より後に作成されたサーバーレスリソースグループを選択する必要があります。リソースグループが 2024 年 6 月 28 日より前に作成された場合は、チケットを送信 してリソースグループをアップグレードしてください。 説明 リリースノートの詳細については、「リリースレコード」をご参照ください。
   入力パラメーターと出力パラメーター	入力パラメーター	[作成] をクリックして、入力パラメーターを追加します。 [作成] をクリックします。パラメーター名: inputs。 値のソース: ドロップダウンリストから MaxComputeDataQuery ノードの outputs パラメーターを選択します。

2. プッシュするデータを設定します。

   • 宛先: [宛先] ドロップダウンリストから宛先を選択します。使用可能な宛先がない場合は、[宛先の作成] をクリックして宛先を作成します。次の表に、宛先を作成するためのパラメーターを示します。

     パラメーター	説明
     タイプ	プッシュチャネルを選択します。DingTalk、Lark、WeCom、Teams がサポートされています。
     名前	ビジネス要件に基づいて名前を入力します。
     Webhook	選択したプッシュチャネルの Webhook URL を入力します。対応するプラットフォームで関連するプッシュチャネルの URL を取得する必要があります。 説明 Lark の Webhook URL の取得方法については、「Webhook トリガーを使用する」をご参照ください。 Teams の Webhook URL の取得方法については、「Microsoft Teams 用のワークフローで受信 Webhook を作成する」をご参照ください。

   • タイトル: MaxCompute データ と入力します。

   • 本文: ビジネス要件に基づいてコンテンツを設定します。詳細については、「データプッシュ」トピックの「プッシュするコンテンツを設定する」セクションをご参照ください。

     説明

     本文 セクションでは、代入ノードによってクエリされたフィールド名をプレースホルダーとして直接使用して、データプッシュノードの入力パラメーターを取得できます。

3. トップツールバーの アイコンをクリックして、設定を保存します。

ステップ 5: フローをテストする

データプッシュフローを設定したら、ワークフローをコミットしてワークフロー内のノードをデプロイする前に、フローをテストする必要があります。

1. DataPushDemo ワークフローをダブルクリックします。

2. MaxComputeDataSync ノードを右クリックし、[現在のノードとその子孫ノードを実行] を選択します。

   説明

   データプッシュフロー内のノードでエラーが発生した場合は、ノードを右クリックして [ログの表示] を選択し、ログを表示します。