如何通过DataWorks加工采集至MaxCompute的数据_大数据开发治理平台 DataWorks(DataWorks)-阿里云帮助中心

本文为您介绍如何通过DataWorks加工采集至MaxCompute的数据，并获取清洗后的数据。

前提条件

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

新建表

进入数据开发页面。
登录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;

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

设计业务流程

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

进入业务流程开发面板，并向面板中拖入两个ODPS SQL节点，依次命名为数据清洗和数据汇聚，并配置如下图所示的依赖关系。

配置ODPS SQL节点

配置数据清洗节点。

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

编写处理逻辑。编写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

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。

编写并执行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=业务日期;