DataWorks:データプッシュノードの設定

ステップ 1：プッシュワークフローの構築

Data_Push_Demo ワークフローをダブルクリックしてワークフローキャンバスを開きます。ノードを順番に接続して、下の図に示すような条件付きデータプッシュワークフローを作成します。このワークフローは、Query_Last_Month_Sales_Total ノードから Check_Sales_Target ノードにデータをルーティングします。Check_Sales_Target ノードは、フローを「準拠」ブランチ（Last_Month_Sales_Target_Met ノードと Push_Top_3_Categories_Target_Met ノードを含む）と「非準拠」ブランチ（Last_Month_Sales_Target_Not_Met ノードと Push_Bottom_3_Categories_Target_Not_Met ノードを含む）に分割します。

ステップ 2：SQL クエリノードの設定

SQL クエリノードを設定して、テストデータをクエリします。 クエリ結果は、ノードコンテキストのパラメーターを使用して、出力パラメーター (outputs) として下流の分岐ノードに渡されます。

1. Query_Last_Month_Sales_Total ノードをダブルクリックし、次の SQL コードを入力します。

   -- 前月の総売上を計算
   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. SQL コードを入力したら、右側にあるスケジューリング設定をクリックします。 [プロパティ] ペインで、次のパラメーターを設定します。

   • スケジュール時刻: 08:00。

   • リソースグループのスケジューリング: サーバーレスリソースグループを選択します。

   • 上流ノード: ワークスペースのルート頂点を使用する を選択します。

   • このノードのレスポンスパラメーター: ノードコンテキストのパラメーター の下で、このノードのレスポンスパラメーター の横にある 代入パラメーターの追加 をクリックして、クエリ結果を下流ノードに渡すための出力パラメーターを追加します。

ステップ 3：分岐ノードの設定

ブランチノードでは、ノードコンテキストのパラメーター 内で、このノードの入力パラメーター を使用して上流の SQL クエリノードから outputs 出力パラメーターを取得します。ノードはその後、条件判定を実行し、このノードのレスポンスパラメーター を使用して条件を満たすデータを出力します。

1. Check_Sales_Target ノードをダブルクリックします。ブランチの追加 をクリックして、「準拠」用と「非準拠」用の 2 つのブランチを作成し、次のように設定します。

   パラメーター	準拠ブランチ	非準拠ブランチ
   ブランチ条件	${inputs[0][0]}>=500000	${inputs[0][0]}<500000
   ノード出力への関連付け	Compliant	Non-compliant
   ブランチの説明	先月の総売上は準拠しています。	先月の総売上は準拠していません。

   説明

   式 [0][0] は、条件の特定のデータポイントを特定するための 2 次元配列参照です。

   • 上流ノードが SQL クエリノードの場合、2 次元配列を使用してパラメーター内の条件データを特定します。

   • 上流ノードが Python ノードの場合、1 次元配列を使用します。

2. Condition Check ノードをダブルクリックして開きます。右側で スケジューリング設定 をクリックし、表示されるダイアログボックスで次の設定項目を設定します。

   パラメータータイプ		説明
   時間プロパティ	スケジューリング周期	日
   	スケジュール時刻	08:00
   	再実行プロパティ	成功および失敗した実行後の再実行が許可されます。
   リソースプロパティ	リソースグループのスケジューリング	サーバーレスリソースグループを選択します。 説明 データプッシュノードを初めて使用する場合、スケジューリングリソースグループをアップグレードするためにチケットを提出する必要があります。
   このノードの出力名		ブランチを設定すると、出力名は自動的に解析され、ここに追加されます。出力名 は、ブランチに定義されている ノード出力への関連付け と一致する必要があります。
   ノードコンテキストのパラメーター	このノードの入力パラメーター	パラメーター名: inputs 値のソース：上流ノード Query last month's total sales から、出力パラメーター outputs を選択します。
   	このノードのレスポンスパラメーター	システムはデフォルトで outputs を追加します。

3. 設定後、 アイコンをクリックして Check_Sales_Target ノードを保存します。

ステップ 4：分岐 SQL ノードの設定

このチュートリアルでは、ブランチ SQL クエリノードは 2 つのノード、Last_Month_Sales_Target_Met_Data と Last_Month_Sales_Target_Not_Met_Data に分割されます。ブランチノード のコンテキストパラメーターを Last_Month_Sales_Target_Met_Data ノードと Last_Month_Sales_Target_Not_Met_Data ノードに出力する必要があります。次に、このノードのレスポンスパラメーター の ノードコンテキストのパラメーター を使用して、ブランチノードからの SQL クエリ結果を、対応する target-met および target-not-met の データプッシュ ノードに出力します。

1. Last_Month_Sales_Target_Met ノードと Last_Month_Sales_Target_Not_Met ノードをダブルクリックしてエディターを開き、それぞれに次の SQL コードを入力します。

   先月の売上目標達成

   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;

   先月の売上目標未達成

   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. SQL コードを入力したら、右側のスケジューリング設定をクリックします。プロパティペインで、以下のパラメーターを設定します。

   • スケジュール時刻: 08:00。

   • リソースグループのスケジューリング：サーバーレスリソースグループを選択します。

   • 依存する先祖ノード：ワークフローは ステップ 1: プッシュワークフローの構築 で作成されているため、上流依存関係が正しいことを確認してください。

     • Last_Month_Sales_Target_Met ノードでは、上流ノードの出力名は [Compliant] である必要があります。

     • Last_Month_Sales_Target_Not_Met ノードについては、上流ノードの出力名は [非準拠] である必要があります。

   • このノードのレスポンスパラメーター: ノードコンテキストのパラメーター で、このノードのレスポンスパラメーター の横にある 代入パラメーターの追加 をクリックして、出力パラメーターを追加します。

3. アイコンをクリックして Last_Month_Sales_Target_Met と Last_Month_Sales_Target_Not_Met ノードを保存します。

ステップ 5：分岐データプッシュノードの設定

2 つのデータプッシュノードを作成する必要があります。各ノードで、ノードコンテキストのパラメーター で このノードの入力パラメーター を設定し、Last_Month_Sales_Target_Met_Data および Last_Month_Sales_Target_Not_Met_Data ブランチ SQL クエリノードからそれぞれ outputs パラメーターを取得します。次に、これらのパラメーターを Body で使用し、それぞれのターゲットにプッシュします。

1. Push_Top_3_Categories_Target_Met ノードと Push_Bottom_3_Categories_Target_Not_Met ノードをダブルクリックします。 ノードエディターで、スケジューリング設定 をクリックし、スケジューリング設定 ペインで以下を設定します。

   パラメータータイプ		値	図解
   スケジューリングパラメータ	パラメーター名	curdate
   	パラメーター値	$[yyyymmddhh:mi:ss]
   時間プロパティ	スケジューリング周期	日
   	スケジュール時刻	08:00 説明 このチュートリアルでは、データが 08:00 に送信先にプッシュされることを保証するために 08:00 を例として使用します。必要に応じて別の時間を設定できます。
   	再実行プロパティ	成功および失敗した実行後の再実行が許可されます。
   リソースプロパティ	リソースグループのスケジューリング	サーバーレスリソースグループを選択します。 説明 データプッシュノードを初めて使用する場合、スケジューリングリソースグループをアップグレードするためにチケットを提出する必要があります。
   ノードコンテキストのパラメーター	このノードの入力パラメーター	入力パラメーターを追加するには、追加 をクリックします: パラメーター名：inputs。 値のソース： Push Top 3 Sales Goal Categories ノードで、Last Month's Sales Goal Data ノードの出力パラメーター outputs を選択します。 Push Bottom Three Underperforming Categories ノードで、Last Month's Underperforming Sales Data ノードから出力パラメーター outputs を選択します。	Push_Top_3_Categories_Target_Met ノード Push_Bottom_3_Categories_Target_Not_Met ノード

2. スケジューリング設定 を設定した後、次のようにデータプッシュコンテンツを設定します。

   • データプッシュ対象: データプッシュ対象 ドロップダウンリストから、送信先を選択します。送信先がリストにない場合は、リストの下部にある データプッシュオブジェクトの作成 をクリックして作成します。

   パラメーター	説明
   タイプ	DingTalk、Lark、WeCom、Microsoft Teams、および メール をサポートしています。
   オブジェクト名	ビジネスニーズに基づいてカスタム名を入力します。
   Webhook	DingTalk、Lark、WeCom、または Microsoft Teams のチャットボットの Webhook URL、またはメールの SMTP の詳細を、それぞれのプラットフォームから取得します。 説明 Lark ボットの Webhook を取得する方法については、「Lark Webhook トリガーの設定」をご参照ください。 Teams Webhook を取得する方法については、「Microsoft Teams ワークフローを使用して受信 Webhook を作成する」をご参照ください。

   • タイトル： Push_Top_3_Categories_Target_Met および Push_Bottom_3_Categories_Target_Not_Met。

   • 本文：必要に応じて設定してください。詳細については、「プッシュコンテンツを設定する」をご参照ください。

     説明

     本文では、上流の SQL クエリノードからの入力パラメーターを、返されたフィールド名をプレースホルダーとして使用して参照できます。

     • Push_Top_3_Categories_Target_Met の例

     • Push_Bottom_3_Categories_Target_Not_Met の例

3. 設定後、 アイコンをクリックして Push_Top_3_Categories_Target_Met と Push_Bottom_3_Categories_Target_Not_Met ノードを保存します。

ステップ 6：条件付きプッシュワークフローのテスト

条件付きプッシュワークフローを設定した後、送信してデプロイする前にテストします。

1. Data_Push_Demo ワークフローページをダブルクリックして開きます。

2. Query Last Month's Total Sales ノードを選択し、右クリックして「現在のノードとその下流ノードを実行」を選択し、実行が完了するのを待ちます。

   説明

   タスクが失敗した場合は、失敗したノードを右クリックしてログを表示を選択すると、ログが表示されます。

ステップ 1：ワークフローの構築

DataPushDemo ワークフローをダブルクリックしてワークフローキャンバスを開きます。SalesAmountPreWeek、Top3CategoryList、および Top3CategoriesPreWeek ノードを順番に接続して、スクリプトデータプッシュワークフローを作成します。

ステップ 2：SQL クエリノードの設定

SQL クエリノードを設定してテストデータをクエリし、ノードコンテキストのパラメーター を使用して、SQL クエリノードの出力パラメーター outputs を生成できます。これにより、SQL クエリの結果を代入ノードに渡すことができます。

1. SalesAmountPreWeek ノードをダブルクリックしてエディターを開き、次の SQL コードを入力します。

   -- 前週の売上上位 3 カテゴリをクエリ
   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. SQL コードを入力した後、右側にあるスケジューリング設定をクリックします。プロパティペインで、次のパラメーターを設定します。

   • スケジュール時刻: 08:00。

   • リソースグループのスケジューリング：サーバーレスリソースグループを選択します。

   • 上流ノード：ワークスペースのルート頂点を使用する を選択します。

   • このノードのレスポンスパラメーター: ノードコンテキストのパラメーター で、このノードのレスポンスパラメーター の横にある 代入パラメーターの追加 をクリックし、クエリ結果を子孫ノードに渡す出力パラメーターを追加します。

3. Query last week's sales data ノードで 保存をクリックします。

ステップ 3：代入ノードの設定

代入ノードは、ノードコンテキストのパラメーター にある このノードの入力パラメーター を使用して上流の SQL クエリノードからoutputs パラメーターを受け取り、このデータを処理し、このノードのレスポンスパラメーター の新しい値を生成してデータプッシュノードに渡します。

1. Top3CategoryList 代入ノードをダブルクリックします。ノードエディターで、次の Python コードを入力します。

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

2. コードを入力した後、右側にあるスケジューリング設定をクリックします。表示されたペインで、次のパラメーターを設定します。

   • スケジュール時刻:08:00。

   • リソースグループのスケジューリング：スケジューリングリソースグループを選択します。

   • ノードコンテキストのパラメーター。

     • このノードの入力パラメーター：

       • パラメーター名：inputs。

       • 値のソース: 上流のSalesAmountPreWeek ノードから outputs パラメーターを選択します。

     • このノードのレスポンスパラメーター: システムによって自動的に生成されます。

3. をクリックして Top 3 Sales in Organization List 代入ノードを保存します。

ステップ 4：データプッシュノードの設定

上流の代入ノードから outputs パラメーターを受信するには、ノードコンテキストのパラメーター で このノードの入力パラメーター を設定します。その後、メッセージ本文でこのパラメーターを使用し、送信先にプッシュします。

1. データプッシュノード Push Top 3 Categories of Last Week をダブルクリックし、スケジューリング設定 をクリックし、スケジューリング設定 パネルで、次のように設定を構成します。

   パラメータータイプ		パラメーター内容	図解
   スケジューリングパラメータ	パラメーター名	curdate。
   	パラメーター値	$[yyyymmddhh:mi:ss]。
   時間プロパティ	スケジューリング周期	毎日。
   	スケジュール時刻	08:00。 説明 このチュートリアルでは、データが 08:00 に送信先にプッシュされることを保証するために 08:00 を例として使用します。必要に応じて別の時間を設定できます。
   	再実行プロパティ	ノードは成功または失敗した後に再実行できます。
   リソースプロパティ	リソースグループのスケジューリング	スケジューリングリソースグループを選択します。 説明 データプッシュノードを初めて使用する場合、スケジューリングリソースグループをアップグレードするためにチケットを提出する必要があります。
   ノードコンテキストのパラメーター	このノードの入力パラメーター	入力パラメーターを追加するには、追加 をクリックします: パラメーター名：inputs。 値のソース：上流ノード Top 3 sales categories last week から出力パラメーター outputs を選択します。

2. スケジューリング設定 を設定した後、次のようにデータプッシュのコンテンツを設定します。

   • データプッシュ対象：データプッシュ対象 ドロップダウンリストから送信先を選択します。 送信先がリストにない場合は、リストの下部にある データプッシュオブジェクトの作成 をクリックして追加します。

   パラメーター	説明
   タイプ	DingTalk、Lark、WeCom、Teams、およびメールをサポートしています。
   オブジェクト名	ビジネスニーズに基づいてカスタム名を指定します。
   Webhook	DingTalk、Lark、WeCom、または Teams のボットの Webhook URL、およびメールの SMTP サーバーの詳細を、それぞれのプラットフォームから取得します。 説明 Lark ボットの Webhook を取得する方法については、「Lark Webhook トリガーの設定」をご参照ください。 Teams Webhook を取得する方法については、「Microsoft Teams ワークフローを使用して受信 Webhook を作成する」をご参照ください。

   • タイトル:先週の売上上位 3 カテゴリー。

   • 本文: 必要に応じてメッセージ本文を設定します。詳細については、「プッシュコンテンツを設定する」をご参照ください。

     説明

     本文では、上流の SQL クエリノードからの入力パラメーターを、返されたフィールド名をプレースホルダーとして使用して参照できます。

     

3. 設定が完了したら、 をクリックして Push Top Three Categories of Last Week プッシュノードを保存します。

ステップ 5：スクリプトデータプッシュワークフローのテスト

スクリプトデータプッシュワークフローを設定した後、送信してデプロイする前にテストします。

1. DataPushDemo ワークフローをダブルクリックしてワークフローキャンバスを開きます。

2. SalesAmountPreWeek ノードを右クリックし、ノードおよび下流の実行 を選択します。プロセスが完了するまで待ちます。

   説明

   タスクが失敗した場合は、失敗したノードを右クリックし、ログを表示 を選択してログを確認します。

   

ステップ 1：プッシュワークフローの構築

Data_Push_Demo_Workflow をダブルクリックします。ワークフローキャンバスには、複数の既存ノードが表示されます。MySQL クエリノード Query_Yesterday_Sales_Total を上流ノードとしてドラッグし、下流のデータプッシュノード Push_Yesterday_Sales_Total に接続します。これにより、次の図に示すようなシンプルなデータプッシュワークフローが作成されます。

ステップ 2：SQL クエリノードの設定

ノードコンテキストのパラメーター を使用して、SQL クエリの結果をデータプッシュノードに渡す outputs パラメーターを生成します。

1. Query_Yesterday_Sales_Total ノードをダブルクリックし、次の SQL コードを入力します。

   -- 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. SQL コードを入力したら、右側のスケジューリング設定をクリックします。プロパティペインで、次のパラメーターを設定します。

   • スケジュール時刻： 08:00。

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

   • 上流ノード：ワークスペースのルート頂点を使用する を選択します。

   • このノードのレスポンスパラメーター: ノードコンテキストのパラメーター セクションで、このノードのレスポンスパラメーター の横にある 代入パラメーターの追加 をクリックして出力パラメーターを追加します。 すると、下流ノードがその値にアクセスできるようになります。

3. 保存をクリックして Query_Yesterday_Sales_Total ノードを保存します。

ステップ 3：データプッシュノードの設定

上流の SQL クエリノードから outputs パラメーターを受け取るように、ノードコンテキストのパラメーター セクションで このノードの入力パラメーター を設定します。その後、このパラメーターをメッセージ本文で使用して、送信先にプッシュできます。

1. Push_Yesterday_Sales_Total データプッシュノードをダブルクリックします。 ノードエディターで、右側の スケジューリング設定 をクリックします。 スケジューリング設定 ペインで、次のパラメーターを設定します。

   パラメータータイプ		詳細	図解
   スケジューリングパラメータ	パラメーター名	curdate
   	パラメーター値	$[yyyymmddhh:mi:ss]
   時間プロパティ	スケジューリング周期	毎日
   	スケジュール時刻	08:00 説明 このチュートリアルでは、データが 08:00 に送信先にプッシュされることを保証するために 08:00 を例として使用します。必要に応じて別の時間を設定できます。
   	再実行プロパティ	ノードは成功または失敗した実行後に再実行できます。
   リソースプロパティ	リソースグループのスケジューリング	既存のスケジューリングリソースグループを選択します。 説明 データプッシュノードを初めて使用する場合、スケジューリングリソースグループをアップグレードするためにチケットを提出する必要があります。
   ノードコンテキストのパラメーター	このノードの入力パラメーター	入力パラメーターを追加するには、追加 をクリックします: パラメーター名： inputs。 値のソース：上流の Query_Yesterday_Sales_Total ノードから outputs パラメーターを選択します。

2. スケジューリング設定 を設定した後、データプッシュコンテンツを次のように設定します。

   • データプッシュ対象: データプッシュ対象 ドロップダウンリストから送信先を選択します。 送信先がリストに表示されていない場合は、データプッシュオブジェクトの作成 をクリックします。

     パラメーター	説明
     タイプ	サポートされているタイプには、DingTalk、Lark、WeCom、Teams、およびメールが含まれます。
     オブジェクト名	ビジネスニーズに基づいてカスタム名を入力します。
     Webhook	DingTalk、Lark、または WeCom のボット、または Microsoft Teams ワークフローの Webhook URL を、それぞれのプラットフォームから取得します。メールの場合は、SMTP の詳細を取得します。 説明 Lark ボットの Webhook を取得する方法については、「Lark Webhook トリガーの設定」をご参照ください。 Teams Webhook を取得する方法については、「Microsoft Teams ワークフローを使用して受信 Webhook を作成する」をご参照ください。

   • タイトル: Push_Yesterday_Sales_Total.

   • 本文: 必要に応じてメッセージ本文を設定します。詳細については、「プッシュコンテンツを設定する」をご参照ください。

     説明

     本文では、上流の SQL クエリノードからの入力パラメーターを、返されたフィールド名をプレースホルダーとして使用して参照できます。

     

3. 保存をクリックして Push_Yesterday_Sales_Total データプッシュノードを保存します。

ステップ 4：シンプルプッシュワークフローのテスト

シンプルなデータプッシュワークフローを設定した後、デプロイのために送信する前にテストします。

1. Data_Push_Demo_Workflow をダブルクリックしてワークフローキャンバスを開きます。

2. Query_Yesterday_Sales_Total ノードを選択して右クリックし、[現在のノードとその下流ノードを実行] を選択します。実行が完了するまで待ちます。

   説明

   タスクが失敗した場合は、失敗したノードを右クリックし、ログを表示を選択すると、ログが表示されます。

   

ステップ 1：ワークフローの構築

Data Push Demo Flow ワークフローキャンバスをダブルクリックして開きます。キャンバス上で、上流ノード Query Yesterday's Total Sales と Query Yesterday's Sales Growth を子孫ノード Push Yesterday's Total Sales and Growth にドラッグして接続し、次の図に示すような結合データプッシュワークフローを作成します。

ステップ 2：SQL クエリノードの設定

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

1. Query Yesterday's Sales Growth ノードをクリックし、次のクエリ SQL コードを記述します。

   -- 一昨日のデータを格納するための一時テーブル 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;
   
   -- 昨日の売上成長を格納するための結果一時テーブルを作成
   CREATE TEMPORARY TABLE IF NOT EXISTS result (
     category VARCHAR(255),
     diff DOUBLE
   );
   -- 成長データを結果テーブルに書き込む
   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;
   
   -- 結果一時テーブルから成長データをクエリ
   SELECT category, diff FROM result;

2. SQL コードを入力したら、右側のスケジューリング設定をクリックします。プロパティペインで、次のパラメーターを設定します。

   • スケジュール時刻: 08:00。

   • リソースグループのスケジューリング：既存のサーバーレスリソースグループを選択します。

   • 上流ノード: ワークスペースのルート頂点を使用する を選択します。

   • このノードのレスポンスパラメーター: ノードコンテキストのパラメーター セクションで、このノードのレスポンスパラメーター の横にある 代入パラメーターの追加 をクリックし、子孫ノードの入力パラメーターの値として出力パラメーターを追加します。

3. 保存をクリックして Query Yesterday's Sales Growth ノードを保存します。

ステップ 3：データプッシュノードの設定

ノードコンテキストのパラメーター の このノードの入力パラメーター を使用して、上流の Query Yesterday's Total Sales ノードと Query Yesterday's Sales Growth ノードから outputs 出力パラメーターを受信することで、Body でパラメーターを使用してターゲットにプッシュできるようになります。

1. Push Yesterday's Total Sales and Growth という名前のデータプッシュノードをダブルクリックします。ノードページが開いたら、スケジューリング設定 をクリックし、スケジューリング設定 パネルで以下の設定項目を設定します。

   パラメータータイプ		説明	例
   スケジューリングパラメータ	パラメーター名	curdate
   	パラメーター値	$[yyyymmddhh:mi:ss]
   時間プロパティ	スケジューリング周期	日
   	スケジュール時刻	08:00 説明 このチュートリアルでは、データが 08:00 に送信先にプッシュされることを保証するために 08:00 を例として使用します。必要に応じて別の時間を設定できます。
   	再実行プロパティ	ノードは、前回の実行が成功したか失敗したかに関わらず再実行できます。
   リソースプロパティ	リソースグループのスケジューリング	既存のリソースグループを選択します。 説明 データプッシュノードを初めて使用する場合、スケジューリングリソースグループをアップグレードするためにチケットを提出する必要があります。
   ノードコンテキストのパラメーター	このノードの入力パラメーター	追加 をクリックして、このノードの入力パラメーターを追加します: パラメーター 1： パラメーター名：inputs1。 値のソース: 上流ノード 昨日の総売上クエリ の出力パラメーター outputs を選択します。 パラメーター 2： パラメーター名： inputs2。 値のソース: 上流ノード 昨日の売上成長のクエリ から出力パラメーター outputs を選択します。

2. スケジューリング設定 を設定した後、次のようにデータプッシュコンテンツを設定します。

   • データプッシュ対象: データプッシュ対象 ドロップダウンリストから送信先を選択します。目的の送信先がない場合は、ドロップダウンリストの右下隅にある データプッシュオブジェクトの作成 をクリックして新しい送信先を作成します。

   パラメーター	説明
   タイプ	サポートされているタイプには、DingTalk、Lark、WeCom、Microsoft Teams、およびメールが含まれます。
   オブジェクト名	ビジネス要件に基づいてカスタム名を指定します。
   Webhook	DingTalk、Lark、WeCom、または Microsoft Teams のチャットボットの Webhook URL、またはメールの SMTP の詳細を、それぞれのプラットフォームから取得します。 説明 Lark ボットの Webhook を取得する方法については、「Lark Webhook トリガーの設定」をご参照ください。 Teams Webhook を取得する方法については、「Microsoft Teams ワークフローを使用して受信 Webhook を作成する」をご参照ください。

   • タイトル: 昨日の総売上高と成長率のプッシュ

   • 本文: 必要に応じて設定します。詳細については、「プッシュコンテンツを設定する」をご参照ください。

     説明

     本文では、上流の SQL クエリノードからの入力パラメーターを、返されたフィールド名をプレースホルダーとして使用して参照できます。

3. 設定が完了したら、 保存をクリックして Push Yesterday's Total Sales and Growth プッシュノードを保存します。

ステップ 4：結合プッシュワークフローのテスト

結合プッシュワークフローをテストする前に、シンプルプッシュワークフローの設定を完了する必要があります。この設定は、後続のデプロイメントの前提条件です。

1. Data Push Demo Flow をダブルクリックして、ビジネスプロセス図ページを開きます。

2. Push Yesterday's Total Sales and Growth ノードを選択し、右クリックして このノードまで実行 を選択し、実行が完了するまで待ちます。

   説明

   タスクが失敗した場合、失敗したノードを右クリックし、ログを表示 を選択してログを表示します。

   

ステップ 1：データプッシュフローの構築

Data Push Demo Flow をダブルクリックします。ワークフローキャンバスで、作成されたノードを表示できます。オフライン同期 ノードをドラッグして、代入ノード の上流ノードとして配置します。Sync MySQL data to ODPS、Query ODPS data、および Push ODPS data の各ノードを順番に接続します。

ステップ 2：バッチ同期ノードの設定

Synchronize MySQL data to ODPS ノードを設定して、データの準備フェーズで MySQL に書き込まれたテストデータを ODPS データソースに同期し、後続の使用に備えます。

1. Synchronize data from MySQL to ODPS ノードをダブルクリックして、バッチ同期ノードを設定します。

   パラメーター		設定	例
   ソース	データソース	MySQL
   	データソース名	作成した MySQL データソースを選択します。
   マイリソースグループ		サーバーレスリソースグループを選択します。
   宛先	データ宛先	MaxCompute (ODPS)
   	データソース名	ワークスペースにバインドされている MaxCompute データソースを選択します。

   設定を完了すると、システムはデータソースとリソースグループ間の接続性を自動的にテストします。テストが成功したら、「次へ」をクリックしてソースと送信先の詳細を設定します。

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

   パラメーター		設定	例
   データソース	データソース	デフォルト設定を維持します： MySQL。 作成した MySQL データソースを選択します。
   	テーブル	orders テーブルを選択します。
   	データフィルタリング	必要に応じて設定できます。このチュートリアルでは、空白のままにします。
   	シャードキー	ソーステーブルの列を分割キーとして使用できます。プライマリキーまたはインデックス付きの列を使用することを推奨します。
   	データプレビュー	MySQL データソースから取得したデータをプレビューして、期待どおりであることを確認します。
   データ宛先	データソース	デフォルト設定を維持します： MaxCompute (ODPS)。 ワークスペースにバインドされている MaxCompute データソースを選択します。
   	Tunnel リソースグループ	これはトンネルクォータを指定します。デフォルトは「パブリック転送リソース」で、MaxCompute の無料クォータです。
   	テーブル	[送信先テーブルスキーマの生成] をクリックして、送信先テーブルを作成します。
   	パーティション情報	毎日増分データをロードするには、日付ベースのパーティションを設定します。たとえば、pt パーティションを ${bizdate} に設定します。
   	書き込みモード	消去 書き込み前の既存データ (挿入上書き)

3. ソースと送信先を設定すると、フィールドを同名マッピングすることができます。

4. チャネル制御を設定します。

   • タスクの希望最大同時実行数: リソースの制限またはタスクの性質により、ランタイムでの実際の同時実行数はこの値以下になる場合があります。実際の同時実行数に基づいて課金されます。このチュートリアルでは、同時実行数 2 で十分です。

   • 同期速度: 速度制限を使用すると、ソースまたは送信先の読み取り/書き込みロードを保護できます。速度制限を設定しない場合、システムは現在のハードウェアがサポートする最大の転送パフォーマンスを提供します。このチュートリアルでは、速度制限は設定しません。

   • ダーティデータポリシー: ダーティデータは許容されません。

   • 分散処理能力: この機能はデフォルトで無効になっています。有効にするには、8 以上の同時実行数が必要です。

   

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

   統合タスクを設定した後、ページの右側にある **[プロパティ]** をクリックします。**[プロパティ]** パネルで、次のパラメーターを設定します。

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

     • パラメーター名: bizdate

     • パラメーター値： $[yyyymmdd-1]

   • スケジュール時刻： 08:00

   • 再実行プロパティ: ドロップダウンリストから、「実行ステータスにかかわらず許可」を選択します。

   • リソースグループのスケジューリング: 作成したスケジューリングリソースグループを選択します。

   • スケジュール依存: バッチ同期タスクの上流依存関係として、ワークスペースのルートノードを選択します。

6. 保存をクリックして Synchronize MySQL data to ODPS バッチ同期ノードを保存します。

ステップ 3：代入ノードの設定

ODPS データソースでは、ノードコンテキストのパラメーター を使用して ODPS SQL ノードでデータをクエリし、そのデータをデータプッシュノードに出力することはサポートされていません。代入ノードを使用して ODPS データをクエリし、その後 ノードコンテキストのパラメーター を使用してデータをデータプッシュノードに出力する必要があります。

1. Query ODPS data ノードをダブルクリックして、その設定ページを開きます。

   1. エディター上部の [言語] ドロップダウンリストから、**[ODPS SQL]** を選択します。

   2. ODPS SQL コードを入力します。

      -- 各パーティション (pt) 内での売上の降順を計算するためのランキング付きサブクエリを作成します。
      -- 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 列の各一意の値に対してランキングが個別に行われることを指定します。
         -- ORDER BY sales DESC は、各パーティション内の行を売上の降順でソートします。
         -- '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  -- 各 pt パーティション内で売上ランクを計算
          FROM orders
          WHERE pt = '${bizdate}'  -- 指定された業務日付のレコードをフィルタリング
      ) AS ranked_orders
      WHERE rank <= 3  -- 各 pt パーティションの売上上位 3 レコードのみを保持

2. SQL コードを入力した後、右側の スケジューリング設定 をクリックします。プロパティ ペインで、次のパラメーターを設定します。

   • スケジュール時刻： 08:00。

   • リソースグループのスケジューリング: 作成したサーバーレスリソースグループを選択します。

   • 親ノード: 親ノードが MySQL to ODPS バッチ同期ノードであるかどうかを確認します。

   • ノードコンテキストのパラメーター: 出力パラメーターの後に追加 をクリックして、子孫ノードの入力パラメーターの値として出力パラメーターを追加します。

     

3. 保存をクリックして ODPS data query ノードを保存します。

ステップ 4：データプッシュノードの設定

ノードコンテキストのパラメーター の このノードの入力パラメーター を設定して、上流の代入ノードから outputs パラメーターを取得し、これらのパラメーターをボディで使用してターゲットにプッシュします。

1. ODPS Data Push という名前のデータプッシュノードをダブルクリックし、スケジューリング設定 をクリックして、スケジューリング設定 パネルで以下の設定を構成します。

   パラメータータイプ		設定	例
   スケジューリングパラメータ	パラメーター名	curdate
   	パラメーター値	$[yyyymmddhh:mi:ss]
   時間プロパティ	スケジューリング周期	DD
   	スケジュール時刻	08:00 説明 このチュートリアルでは、データが 08:00 に送信先にプッシュされることを保証するために 08:00 を例として使用します。必要に応じて別の時間を設定できます。
   	再実行プロパティ	実行ステータスに関わらず許可。
   リソースプロパティ	リソースグループのスケジューリング	データプッシュ機能のリリース日以降に作成されたリソースグループを選択します。リソースグループがこの日より前に作成された場合は、チケットを提出してアップグレードしてください。 説明 データプッシュノード機能は 2024 年 6 月 28 日にリリースされました。DataWorks のその他のリリースノートについては、「リリース履歴」をご参照ください。
   ノードコンテキストのパラメーター	このノードの入力パラメーター	このノードに入力パラメーターを追加するには、追加をクリックします: パラメーター名： inputs 値のソース: 上流の ODPS Data Query ノードから outputs 出力パラメーターを選択します。

2. スケジューリング設定 を設定した後、次のようにデータプッシュの内容を設定します。

   • データプッシュ対象：データプッシュ対象 ドロップダウンリストから必須の送信先を選択します。送信先が存在しない場合は、ドロップダウンリストの右下にあるデータプッシュオブジェクトの作成をクリックして新しい送信先を作成できます。

   パラメーター	説明
   タイプ	サポートされている送信先には、DingTalk、Lark、WeCom、Microsoft Teams、およびメールが含まれます。
   オブジェクト名	ビジネス要件に基づいてカスタム名を指定します。
   Webhook	DingTalk、Lark、WeCom、または Microsoft Teams のボットの Webhook URL、またはメールの SMTP の詳細を、それぞれのプラットフォームから取得する必要があります。 説明 Lark ボットの Webhook を取得する方法については、「Lark Webhook トリガーの設定」をご参照ください。 Teams Webhook を取得する方法については、「Microsoft Teams ワークフローを使用して受信 Webhook を作成する」をご参照ください。

   • タイトル：ODPS Data Push

   • 本文：必要に応じて設定します。詳細については、「プッシュコンテンツの設定」をご参照ください。

     説明

     メッセージ本文では、上流の代入ノードのクエリ結果のフィールド名を直接プレースホルダーとして使用できます。

3. 設定が完了したら、 保存アイコンをクリックしてデータプッシュノードを保存します。

ステップ 5：MaxCompute データプッシュフローのテスト

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

1. Data_Push_Demo ワークフローページをダブルクリックして開きます。

2. Sync Data from MySQL to ODPS ノードを選択し、右クリックして「現在のノードとその下流ノードを実行」を選択し、実行が完了するのを待ちます。

   説明

   タスクが失敗した場合、失敗したノードを右クリックしてログを表示を選択し、ログを表示します。