DataWorks:Configure data push nodes

Step 1: Build the flow

On the DataPushDemo workflow canvas, connect nodes in this order:

SalesAmountPreMonth → Condition → CompliantData → Top3Categories

and

Condition → NonCompliantData → Bottom3Categories

Step 2: Configure the SQL query node

1. Double-click SalesAmountPreMonth. Write the following query:

-- Query total sales for the previous month
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');

1. On the Properties tab, configure:

   • Scheduled time: 08:00

   • Resource Group: Select a serverless resource group

   • Parent Nodes: Select Add Root Node

   • Output Parameters: In Input and Output Parameters, click Add assignment parameter next to Output Parameters to expose the outputs parameter.

Step 3: Configure the branch node

1. Double-click Condition. In the Definition section, click Add Branch and configure two branches:

   Parameter	Compliant branch	Non-compliant branch
   Condition	${inputs[0][0]}>=500000	${inputs[0][0]}<500000
   Associated node output	Compliant	Non-compliant
   Description	Sales meets threshold	Sales below threshold

2. On the Properties tab, configure:

   Setting	Value
   Scheduling cycle	Day
   Scheduled time	08:00
   Rerun	Allow Regardless of Running Status
   Resource Group	Select an existing resource group for scheduling
   Input Parameters	Parameter Name: inputs; Value Source: outputs of SalesAmountPreMonth
   Output Parameters	Added automatically by the system

3. Save the configuration by clicking the icon.

Step 4: Configure the child nodes

Both CompliantData and NonCompliantData are child nodes of the branch node. Each queries a subset of the data and exposes an outputs parameter for its downstream push node.

1. Double-click CompliantData and write the following query:

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');

-- Temporary table for top-3 categories
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;

SELECT category, sales, all_cat_sales_volume_month FROM temp_array;

1. Double-click NonCompliantData and write the following query:

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');

-- Temporary table for bottom-3 categories
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;

SELECT category, sales, all_cat_sales_volume_month FROM temp_array;

1. For each node, on the Properties tab, configure:

   • Scheduled time: 08:00

   • Resource Group: Select a serverless resource group

   • Parent Nodes: Verify the dependency from the branch node. Check that Output Name of Ancestor Node is Compliant for CompliantData and Non-compliant for NonCompliantData

   • Output Parameters: Click Add assignment parameter to expose the outputs parameter

2. Save each node's configuration.

Step 5: Configure the data push nodes

1. Double-click Top3Categories. On the Properties tab, configure: Screenshots:

   Setting	Value
   Scheduling parameter name	curdate
   Parameter value	$[yyyymmddhh:mi:ss]
   Scheduling cycle	Day
   Scheduled time	08:00
   Rerun	Allow Regardless of Running Status
   Resource Group	Select an existing resource group for scheduling
   Input Parameters	Parameter Name: inputs; Value Source: outputs of CompliantData

2. Apply the same settings to Bottom3Categories, with Input Parameters pointing to outputs of NonCompliantData.

3. For each push node, configure the push destination and content:

   • Destination: Select from the Destination drop-down, or click Create Destination and fill in:

     Parameter	Description
     Type	DingTalk, Lark, WeCom, Microsoft Teams, or Email
     Name	A name for this destination
     Webhook	The webhook URL from the target platform. For Lark, see Configure a Lark webhook trigger. For Teams, see Create incoming webhooks for Microsoft Teams

   • Title:

     • Top3Categories: Categories whose sales amount ranks the top three

     • Bottom3Categories: Categories whose sales amount ranks the bottom three

   • Body: Use column names returned by the SQL query as placeholders. See Configure the push content for placeholder syntax and examples. Sample configurations:

4. Save each node's configuration.

Step 6: Test the flow

1. Double-click the DataPushDemo workflow.

2. Right-click SalesAmountPreMonth and select Run Current Node and Its Descendant Nodes.

If a node fails, right-click it and select View Log to inspect the logs.

Step 1: Build the flow

Connect nodes in this order:

SalesAmountPreWeek → Top3CategoryList → Top3CategoriesPreWeek

Step 2: Configure the SQL query node

1. Double-click SalesAmountPreWeek. Write the following query:

-- Query top-3 categories by total sales for the previous week
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;

1. On the Properties tab, configure:

   • Scheduled time: 08:00

   • Resource Group: Select a serverless resource group

   • Parent Nodes: Select Add Root Node

   • Output Parameters: Click Add assignment parameter to expose the outputs parameter

2. Save the configuration.

Step 3: Configure the assignment node

The assignment node reads the SQL query output and formats it as a human-readable message string. The formatted string becomes the outputs parameter passed to the push node.

1. Double-click Top3CategoryList. Set Language to Python and write the following code:

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()

1. On the Properties tab, configure:

   • Scheduled time: 08:00

   • Resource Group: Select an existing resource group for scheduling

   • Input Parameters: Parameter Name: inputs; Value Source: outputs of SalesAmountPreWeek

   • Output Parameters: Added automatically by the system

2. Save the configuration.

Step 4: Configure the data push node

1. Double-click Top3CategoriesPreWeek. On the Properties tab, configure: Screenshots:

   Setting	Value
   Scheduling parameter name	curdate
   Parameter value	$[yyyymmddhh:mi:ss]
   Scheduling cycle	Day
   Scheduled time	08:00
   Rerun	Allow Regardless of Running Status
   Resource Group	Select an existing resource group for scheduling
   Input Parameters	Parameter Name: inputs; Value Source: outputs of Top3CategoryList

2. Configure the push destination and content:

   • Destination: Select or create a destination (DingTalk, Lark, WeCom, or Teams)

   • Title: Categories whose sales amount ranks the top three in the previous week

   • Body: Use the field names from the assignment node output as placeholders. See Configure the push content for details

3. Save the configuration.

Step 5: Test the flow

1. Double-click the DataPushDemo workflow.

2. Right-click SalesAmountPreWeek and select Run Current Node and Its Descendant Nodes.

If a node fails, right-click it and select View Log to inspect the logs.

Step 1: Build the flow

Connect nodes in this order:

SalesAmountPreDay → PushSalesAmountPreDay

Step 2: Configure the SQL query node

1. Double-click SalesAmountPreDay. Write the following query:

-- Create a temporary table for yesterday's total sales
CREATE TEMPORARY TABLE IF NOT EXISTS temp_array (
  total_amount DOUBLE
);

-- Write yesterday's total sales to the temporary table
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');

SELECT total_amount FROM temp_array;

1. On the Properties tab, configure:

   • Scheduled time: 08:00

   • Resource Group: Select a serverless resource group

   • Parent Nodes: Select Add Root Node

   • Output Parameters: In Input and Output Parameters, click Add assignment parameter next to Output Parameters to expose the outputs parameter

2. Save the configuration.

Step 3: Configure the data push node

1. Double-click PushSalesAmountPreDay. On the Properties tab, configure: Screenshots:

   Setting	Value
   Scheduling parameter name	curdate
   Parameter value	$[yyyymmddhh:mi:ss]
   Scheduling cycle	Day
   Scheduled time	08:00
   Rerun	Allow Regardless of Running Status
   Resource Group	Select an existing resource group for scheduling
   Input Parameters	Parameter Name: inputs; Value Source: outputs of SalesAmountPreDay

2. Configure the push destination and content:

   • Destination: Select or create a destination

   • Title: Total sales amount for yesterday

   • Body: Use total_amount as a placeholder to include the query result. See Configure the push content

3. Save the configuration.

Step 4: Test the flow

1. Double-click the DataPushDemo workflow.

2. Right-click SalesAmountPreDay and select Run Current Node and Its Descendant Nodes.

If a node fails, right-click it and select View Log to inspect the logs.

Step 1: Build the flow

Connect nodes in this order:

SalesAmountPreDay → CombinedPush

SalesGrowthPreDay → CombinedPush

SalesAmountPreDay is shared with the simple push flow.

Step 2: Configure the SQL query node

1. Double-click SalesGrowthPreDay. Write the following query:

-- Collect the day-before-yesterday's sales by category
CREATE TEMPORARY TABLE IF NOT EXISTS temp_array1 (
  category VARCHAR(255),
  sales DOUBLE
);
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;

-- Collect yesterday's sales by category
CREATE TEMPORARY TABLE IF NOT EXISTS temp_array2 (
  category VARCHAR(255),
  sales DOUBLE
);
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;

-- Calculate day-over-day sales growth
CREATE TEMPORARY TABLE IF NOT EXISTS result (
  category VARCHAR(255),
  diff DOUBLE
);
INSERT INTO result (category, diff)
SELECT temp_array2.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;

1. On the Properties tab, configure:

   • Scheduled time: 08:00

   • Resource Group: Select a serverless resource group

   • Parent Nodes: Select Add Root Node

   • Output Parameters: Click Add assignment parameter to expose the outputs parameter

2. Save the configuration.

Step 3: Configure the data push node

1. Double-click CombinedPush. On the Properties tab, configure: Screenshots:

   Setting	Value
   Scheduling parameter name	curdate
   Parameter value	$[yyyymmddhh:mi:ss]
   Scheduling cycle	Day
   Scheduled time	08:00
   Rerun	Allow Regardless of Running Status
   Resource Group	Select an existing resource group for scheduling
   Input parameter 1	Parameter Name: inputs1; Value Source: outputs of SalesAmountPreDay
   Input parameter 2	Parameter Name: inputs2; Value Source: outputs of SalesGrowthPreDay

2. Configure the push destination and content:

   • Destination: Select or create a destination

   • Title: Sales amount and the sales amount growth for yesterday

   • Body: Use column names (total_amount, category, diff) as placeholders. See Configure the push content

3. Save the configuration.

Step 4: Test the flow

1. Double-click the DataPushDemo workflow.

2. Right-click CombinedPush and select Run Current Node and Its Descendant Nodes.

If a node fails, right-click it and select View Log to inspect the logs.

Step 1: Build the flow

Connect nodes in this order:

MaxComputeDataSync → MaxComputeDataQuery → MaxComputeDataPush

Step 2: Configure the batch synchronization node

1. Double-click MaxComputeDataSync. Configure the source and destination: After the connectivity test passes, click Next.

   Section	Parameter	Value
   Source	Source	MySQL
   	Data source name	Select the MySQL data source you added
   	Resource group	Select a serverless resource group
   Destination	Destination	MaxCompute (ODPS)
   	Data source name	Select the MaxCompute data source added to your workspace

   

2. Configure source and destination details: Screenshots:

   Section	Parameter	Value
   Source	Data source	MySQL (retain default); select the MySQL data source
   	Table	orders
   	Data filtering	Leave blank
   	Split key	Use the primary key column or an indexed column
   Destination	Data source	MaxCompute (retain default); select the MaxCompute data source
   	Tunnel resource group	Shared transmission resources
   	Table	Click Generate Destination Table Schema
   	Partition information	Set pt to ${bizdate} for daily incremental partitions
   	Write mode	Clean up existing data before writing (INSERT OVERWRITE)

3. Map source fields to destination fields with matching names.

4. Configure channel control:

   • Task expected maximum concurrency: 2

   • Synchronization rate: No current limit

   • Policy for dirty data records: Disallow Dirty Data Records

   • Distributed execution: Off (turn on only if concurrency >= 8)

5. On the Properties tab, configure scheduling:

   • Scheduling parameter name: bizdate; Parameter value: $[yyyymmdd-1]

   • Scheduled time: 08:00

   • Rerun: Allow Regardless of Running Status

   • Resource Group: Select an existing resource group

   • Dependencies: Select Add Root Node

6. Save the configuration.

Step 3: Configure the assignment node

The assignment node queries the MaxCompute data and exposes the results via outputs.

1. Double-click MaxComputeDataQuery. Set Language to ODPS SQL and write the following query:

-- Use DENSE_RANK() to rank orders by sales within each partition,
-- then return the top-3 orders per partition date.
--
-- PARTITION BY pt: groups rows by the partition date field
-- ORDER BY sales DESC: ranks rows from highest to lowest sales
-- WHERE rank <= 3: keeps only the top-3 rows per partition

SELECT
  order_id,   -- Order ID
  category,   -- Product category
  sales,      -- Sales amount
  datetime,   -- Payment time
  pt          -- Partition date
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}'
) AS ranked_orders
WHERE rank <= 3;

1. On the Properties tab, configure:

   • Scheduled time: 08:00

   • Resource Group: Select a serverless resource group

   • Dependent Upstream Node: Verify that Ancestor Node Name is MaxComputeDataSync

   • Output Parameters: Click Create next to Output Parameters to expose the outputs parameter

2. Save the configuration.

Step 4: Configure the data push node

1. Double-click MaxComputeDataPush. On the Properties tab, configure: Screenshots:

   Setting	Value
   Scheduling parameter name	curdate
   Parameter value	$[yyyymmddhh:mi:ss]
   Scheduling cycle	Day
   Scheduled time	08:00
   Rerun	Allow Regardless of Running Status
   Resource Group	Select a serverless resource group created after June 28, 2024 (the release date of the data push feature). If your resource group was created before this date, submit a ticket to upgrade it. See Release records for details
   Input Parameters	Parameter Name: inputs; Value Source: outputs of MaxComputeDataQuery

2. Configure the push destination and content:

   • Destination: Select or create a destination

   • Title: MaxCompute data

   • Body: Use the column names returned by the assignment node (order_id, category, sales, datetime, pt) as placeholders. See Configure the push content

3. Save the configuration.

Step 5: Test the flow

1. Double-click the DataPushDemo workflow.

2. Right-click MaxComputeDataSync and select Run Current Node and Its Descendant Nodes.

If a node fails, right-click it and select View Log to inspect the logs.