使用ODPSQL节点清洗并汇聚MaxCompute数据-大数据开发治理平台 DataWorks-阿里云-大数据开发治理平台 DataWorks(DataWorks)-阿里云帮助中心

前提条件

开始本文的操作前，请首先完成准备数据中的操作。

新建表

进入数据开发页面。
登录DataWorks控制台，切换至目标地域后，单击左侧导航栏的数据开发与运维 > 数据开发，在下拉框中选择对应工作空间后单击进入数据开发。
在数据开发页面，展开目标业务流程。
右键单击MaxCompute，选择新建表。
在新建表对话框中，输入表名称，单击新建。
重要
表名必须以字母开头，不能包含中文或特殊字符，且不能超过64个字符。
此处需要创建的数据表，如下所示：
- 创建三张表，分别存储同步过来的电量下降趋势数据、指标数据和窃电标志数据清洗之后的数据（clean_trend_data、clean_indicators_data和clean_steal_flag_data）。
- 创建表data4ml，存储汇聚后的数据。

打开创建的表，单击DDL，分别输入以下相应的建表语句。

--清洗后的电量下降趋势数据
CREATE TABLE clean_trend_data (
    uid bigint,
    trend bigint
)
PARTITIONED BY (dt string)
LIFECYCLE 7;

--清洗后的指标数据
CREATE TABLE clean_indicators_data (
    uid bigint,
    xiansun bigint,
    warnindicator bigint
)
COMMENT '*'
PARTITIONED BY (ds string)
LIFECYCLE 36000;

--清洗后的窃电标志数据
CREATE TABLE clean_steal_flag_data (
    uid bigint,
    flag bigint
)
COMMENT '*'
PARTITIONED BY (ds string)
LIFECYCLE 36000;

--汇聚后的数据
CREATE TABLE data4ml (
    uid bigint,
    trend bigint,
    xiansun bigint,
    warnindicator bigint,
    flag bigint
)
COMMENT '*'
PARTITIONED BY (ds string)
LIFECYCLE 36000;

建表语句输入完成后，单击生成表结构并确认覆盖当前操作。
返回建表页面后，在基本属性中输入表的中文名。
完成设置后，分别单击提交到开发环境和提交到生产环境。

表编辑页面顶部工具栏中，依次显示DDL、提交到开发环境、从生产环境加载和提交到生产环境按钮。

设计业务流程

业务流程的新建及依赖关系的配置请参见新建业务流程。

进入业务流程开发面板，并向面板中拖入两个ODPS SQL节点，依次命名为数据清洗和数据汇聚。配置完成后的业务流程DAG显示完整依赖：start → 电量下降趋势数据同步、窃电标志数据同步、指标数据同步（三个并行节点）→ 数据清洗 → 数据汇聚。

配置ODPS SQL节点

配置数据清洗节点。

双击数据清洗节点，进入节点配置页面。

编写处理逻辑。

SQL逻辑如下所示。

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}
;

单击工具栏中的图标。

配置数据汇聚节点。

双击数据汇聚节点，进入节点配置页面。

编写处理逻辑。

SQL逻辑如下所示。

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
;

单击工具栏中的图标。

提交业务流程

打开业务流程配置面板，单击工具栏中的图标。
选择提交对话框中需要提交的节点，输入备注，并选中忽略输入输出不一致的告警。
单击确认，待显示提交成功即可。

运行业务流程

打开业务流程配置面板，单击工具栏中的图标。
在左侧导航栏，单击临时查询。
在临时查询页面，右键单击临时查询，选择新建节点 > ODPS SQL。

左侧导航栏中选中临时查询后，右键展开新建节点子菜单，可选择ODPS SQL、Shell、OceanBase、DB2等多种节点类型。

编写并执行SQL语句，查看导入表clean_trend_data、clean_indicators_data、clean_steal_flag_data和data4ml的记录数。

SQL语句如下所示，其中分区列需要更新为业务日期。例如，任务运行的日期为20190809，则业务日期为20190808。

--查看是否成功写入MaxCompute
SELECT count(*) from clean_trend_data where dt=业务日期;
SELECT count(*) from clean_indicators_data where ds=业务日期;
SELECT count(*) from clean_steal_flag_data where ds=业务日期;
SELECT count(*) from data4ml where ds=业务日期;

发布业务流程

提交业务流程后，表示任务已进入开发环境。由于开发环境的任务不会自动调度，您需要将配置完成的任务发布至生产环境。

说明

将任务发布至生产环境前，您需要对代码进行测试，确保其正确性。

打开业务流程配置面板，单击工具栏中的图标。
在创建发布包页面，选中待发布的任务，单击添加到待发布。
进入右上角的待发布列表，单击全部打包发布。

待发布面板列出5项待发布任务（数据汇聚、数据清洗、窃电标志数据同步、电量下降趋势数据同步、指标数据同步），各项状态均为检查通过，变更类型为新增。
在发布包列表页面查看已发布的内容。

在生产环境运行任务

任务发布成功后，单击右上角的运维中心。
选择周期任务运维 > 周期任务中的相应节点。

在运维中心的周期任务运维 > 周期任务页面，左侧列出所有周期任务节点，勾选目标节点后右侧显示对应的DAG依赖关系图。
右键单击DAG图中的start节点，选择补数据 > 当前节点及下游节点。

在DAG图中右键菜单提供查看节点详情、查看代码、编辑节点、补数据等操作选项，补数据子菜单中包含当前节点、当前节点及下游节点、海量节点模式和高级模式。
选中需要补数据的任务，并选择业务日期。

在新建补数据任务页面，方式选择默认为手动选择，已选任务列表显示start、数据清洗、数据汇聚及三个数据同步节点。在设置补数据运行策略区域，设置业务日期范围，其余参数（调度周期、并行、顺序等）保持默认即可。确认配置后，单击页面底部的提交。
单击提交。
在补数据实例页面，单击刷新，直至SQL任务都运行成功即可。

加工数据