Process data

Create tables

1. Go to the DataStudio page.

   Log on to the DataWorks console. In the target region, click Data Development and O&M > Data Development in the left-side navigation pane. Select a workspace from the drop-down list and click Go to Data Development.

2. On the DataStudio page, expand the target workflow.

3. Right-click Maxcompute and select Create Table.

4. In the Create Table dialog box, enter a table name and click Create.

   Important

   A table name must start with a letter, be 64 characters or less, and cannot contain Chinese characters or special characters.

   In this tutorial, create the following tables:

   • Create three tables to store the cleansed data for the synchronized electricity consumption downward trend, indicators, and electricity theft flags (clean_trend_data, clean_indicators_data, and clean_steal_flag_data, respectively).

   • Create the table data4ml to store the aggregated data.

5. For each table you create, go to its DDL tab and enter the corresponding CREATE TABLE statement from the examples below.

   -- Cleansed power-drop trend data
   CREATE TABLE clean_trend_data (
       uid bigint,
       trend bigint
   )
   PARTITIONED BY (dt string)
   LIFECYCLE 7;

   -- Cleansed indicators data
   CREATE TABLE clean_indicators_data (
       uid bigint,
       xiansun bigint,
       warnindicator bigint
   )
   COMMENT '*'
   PARTITIONED BY (ds string)
   LIFECYCLE 36000;

   -- Cleansed power-theft flag data
   CREATE TABLE clean_steal_flag_data (
       uid bigint,
       flag bigint
   )
   COMMENT '*'
   PARTITIONED BY (ds string)
   LIFECYCLE 36000;

   -- Aggregated data
   CREATE TABLE data4ml (
       uid bigint,
       trend bigint,
       xiansun bigint,
       warnindicator bigint,
       flag bigint
   )
   COMMENT '*'
   PARTITIONED BY (ds string)
   LIFECYCLE 36000;

6. After entering the table creation statement, click "Generate Table Structure" and confirm to overwrite the current operation.

7. After returning to the table creation page, enter the Chinese name for the table in the Basic Properties section.

8. After you complete the settings, click Commit to Development Environment and Commit to Production Environment.

Design a workflow

To create a workflow and configure dependencies, see Design a workflow.

Configure ODPS SQL nodes

• Configure the Data Cleansing node.

  1. Double-click the Data Cleansing node to go to the node configuration page.

  2. Write the processing logic.

     The SQL logic is as follows.

     INSERT OVERWRITE TABLE clean_trend_data PARTITION(dt=${bdp.system.bizdate})
     SELECTuid
     ,trend
     FROMtrend_data
     WHEREtrend IS NOT NULL
     ANDuid != 0
     ANDdt = ${bdp.system.bizdate}
     ;
     
     INSERT OVERWRITE TABLE clean_steal_flag_data PARTITION(ds=${bdp.system.bizdate})
     SELECTuid
     ,flag
     FROMsteal_flag_data
     WHEREuid != 0
     ANDds = ${bdp.system.bizdate}
     ;
     
     INSERT OVERWRITE TABLE clean_indicators_data PARTITION(ds=${bdp.system.bizdate})
     SELECTuid
     ,xiansun,warnindicator
     FROMindicators_data
     WHEREuid != 0
     ANDds = ${bdp.system.bizdate}
     ;

  3. Click the icon in the toolbar.

• Configure the Data Aggregation node.

  1. Double-click the Data Aggregation node to go to the node configuration page.

  2. Write the processing logic.

     The SQL logic is as follows.

     INSERT OVERWRITE TABLE data4ml PARTITION (ds=${bdp.system.bizdate})
     SELECTa.uid
     ,trend
     ,xiansun
     ,warnindicator
     ,flag
     FROM
     (
     SELECT uid,trend FROM clean_trend_data where dt=${bdp.system.bizdate}
     )a
     FULL OUTER JOIN
     (
     SELECT uid,xiansun,warnindicator FROMclean_indicators_data where ds=${bdp.system.bizdate}
     )b
     ONa.uid = b.uid
     FULL OUTER JOIN
     (
     SELECT uid,flag FROMclean_steal_flag_data where ds=${bdp.system.bizdate}
     )c
     ONb.uid = c.uid
     ;

  3. Click the icon in the toolbar.

Submit the workflow

1. Open the workflow configuration panel and click the icon in the toolbar.

2. In the Submission dialog box, select the nodes to submit, enter Remarks, and select Ignore I/O Inconsistency Alerts.

3. Click Confirm. Wait until Submitted successfully is displayed.

Run the workflow

1. Open the workflow configuration panel and click the icon in the toolbar.

2. In the left-side navigation pane, click Ad Hoc Query.

3. On the Ad Hoc Query page, right-click Ad Hoc Query and choose Create Node > ODPS SQL.

4. Write and run SQL statements to check the number of records imported into the tables clean_trend_data, clean_indicators_data, clean_steal_flag_data, and data4ml.

   Use the following SQL statements. Replace the partition column value with the business date. For example, if the task runs on 20190809, the business date is 20190808.

   -- Check whether data is written to MaxCompute
   SELECT count(*) from clean_trend_data where dt=Business date;
   SELECT count(*) from clean_indicators_data where ds=Business date;
   SELECT count(*) from clean_steal_flag_data where ds=Business date;
   SELECT count(*) from data4ml where ds=Business date;

Deploy the workflow

After you submit the workflow, the tasks remain in the development environment and are not automatically scheduled. Deploy the configured tasks to the production environment.

1. Open the workflow configuration panel and click the icon in the toolbar.

2. On the Create Deploy Task page, select the tasks to deploy and click Add to List.

3. Go to Nodes to Deploy in the upper-right corner and click Deploy All.

4. On the Deployment Packages page, view the deployed content.

Run tasks in the production environment

1. After the tasks are deployed, click Operation Center in the upper-right corner.

2. Choose Auto Triggered Task O&M > Auto Triggered Task and select the corresponding node.

3. Right-click the start node in the DAG and choose Backfill Data > Current and Descendant Nodes Retroactively.

4. Select the tasks for which you want to backfill data, and set the Data Timestamp.

5. Click Submission.

6. On the Data replenishment instance page, click Refresh until all SQL tasks are run successfully.