离线数据集成(DataWorks)
物联网平台数据服务中的平台系统表、产品属性时序表、产品事件表和自定义存储表数据,可以集成到阿里云大数据开发治理平台(DataWorks)中构建数据仓库,以提升数据应用效率。本文介绍如何使用DataWorks的数据集成平台集成物联网平台中数据服务的数据。
背景信息
DataWorks的数据集成是稳定高效、弹性伸缩的数据同步平台,致力于提供复杂网络环境下、丰富的异构数据源之间高速稳定的数据移动及同步能力。详细说明,请参见数据集成概述。
使用DataWorks中数据集成的IoT Reader插件可以读取物联网平台企业版实例中数据,IoT Reader支持的数据类型、字段映射和数据源等参数与配置的详细说明,请参见IoT数据源。
使用限制
使用DataWorks集成物联网平台企业版实例中数据的限制如下。
限制项 | 说明 |
地域 | 华东2(上海)、华南1(深圳)、华北2(北京)。 |
物联网平台企业版实例 | 标准型、尊享型。实例类型说明,请参见企业版实例类型说明。 |
集成的物联网平台数据源 | 重要 如果需要将物联网平台的产品属性时序表和产品事件表的数据集成到DataWorks,需先在物联网平台的数据服务中开启该产品的备份。具体操作,请参见备份设备数据源。 |
独享资源组 | DataWorks中数据集成的Reader插件仅支持独享数据集成资源组。详细说明,请参见独享数据集成资源组。 |
读取时序数据 | 支持指定数据的时间范围。 |
数据过滤 | 不支持。 |
计费说明
物联网平台数据集成流出会消耗数据处理单元(CU)。详细说明,请参见数据服务计费说明的数据集成。
DataWorks中数据集成的计费逻辑,请参见计费逻辑说明。
使用流程
前提条件
如果集成产品属性时序表或产品事件表数据,需在物联网平台的数据服务中,开启对应产品的设备数据源备份。具体操作,请参见备份设备数据源。
在DataWorks中配置数据集成的离线同步任务
通过向导模式配置离线同步任务
步骤一:新建离线同步节点:根据界面提示创建离线同步节点。
步骤二:配置同步网络链接:选择离线同步任务的数据来源、用于执行同步任务的我的资源组,以及数据去向,并测试连通性。
重要数据来源选择IoT,我的资源组只支持独享数据集成资源组,数据去向由您自定义。
步骤三:配置数据来源与去向:配置同步任务的数据来源和数据去向的详情信息。
参数
说明
表
数据存储表的表标识符。
导出日期数据(时序表)
日期格式为
yyyyMMdd
,例如20150101
,表示导出该日的数据。步骤四:配置字段映射关系:左侧的源头表字段和右侧的目标表字段为一一对应的关系。
参数
说明
备注
同名映射
单击同名映射,可以根据名称建立相应的映射关系。
重要数据类型需保证一一对应匹配。
同行映射
单击同行映射,可以在同行建立相应的映射关系。
取消映射
单击取消映射,可以取消建立的映射关系。
您可以单击添加一行可以增加单个字段,您也可以将鼠标移动至需要删除的字段上,单击删除图标删除字段。
步骤五:配置通道:通过通道配置,控制数据同步过程相关属性。
通过脚本模式配置离线同步任务
步骤一:新建离线同步节点:根据界面提示创建离线同步节点。
步骤二:配置同步网络链接:选择离线同步任务的数据来源、用于执行同步任务的我的资源组,以及数据去向,并测试连通性。
步骤四:编辑脚本,配置同步任务:编辑JSON脚本,指定IoT Reader以及对应Writer的参数。
以下为一个完整的示例脚本,示例中从IoT Reader数据源的产品时序存储表中读取数据,写入到一张MaxCompute表中。
重要实际场景中,需根据代码中注释修改对应参数值,且在运行时,删除代码中的注释。
{ "type": "job", "version": "2.0", "steps": [ { "stepType": "iot", // 请指定 "iot",表示IoT Reader数据源。 "parameter": { "accessId": "LT********", // 用于访问物联网企业实例的AccessKey ID。 "accessKey": "******", // 用于访问物联网企业实例的AccessKey Secret。 "regionId": "cn-shanghai", // 物联网平台企业版实例的地域ID。 "instanceId": "iot-*******", // 物联网平台企业版实例ID。 "column": [ // 读取IoT数据存储表的列信息。 "product_key", "device_name", "iot_id", "event_time", "event_date", "items" ], "table": "product.h******", // 数据存储表的表标识符。 "date": "${bizdate}" // 日期格式为yyyyMMdd,例如“20151111”,表示导出该日的数据。 }, "name": "Reader", "category": "reader" }, { "stepType": "odps", "parameter": { "partition": "ds=${bizdate}", "truncate": true, "indexes": [], "datasource": "odps_first", "envType": 0, "isSupportThreeModel": false, "column": [ "product_key", "device_name", "iot_id", "event_time", "event_date", "items" ], "emptyAsNull": false, "table": "ods_product_timeline_dd" }, "name": "Writer", "category": "writer" }, { "copies": 1, "parameter": { "nodes": [], "edges": [], "groups": [], "version": "2.0" }, "name": "Processor", "category": "processor" } ], "setting": { "errorLimit": { "record": "" }, "locale": "zh", "speed": { "throttle": false, "concurrent": 2 } }, "order": { "hops": [ { "from": "Reader", "to": "Writer" } ] } }