工业互联网平台综合应用课件：工业互联网平台实时数据开发

工业互联网平台实时数据开发《工业互联网平台综合应用》/1工业设备实时数据开发《工业互联网平台综合应用》/知识准备任务实施目录0102◼数据库定义语言(DDL)：用来定义数据库模式。DDL主要包括数据库模式定义、数据库存储结构、存取方法定义、数据库模式修改和删除功能。◼数据库操纵语言(DML)：用来表示用户对数据库的操作请求

，功能主要包括查询数据库的查询、删除、修改、新增功能◼数据查询语言(DQL)：用来查询数据库中表的记录

，关键字：select，where等◼数据控制语言(DCL)：用来定义数据库的访问权限和安全级别。SQL(Structured

Query

Lanauage)结构化查询语言是关系数据库中最普遍使用的语言。主要包括查询、数据操纵、数据定义、数据控制功能，是一种通用的、功能强大的关系数据库的标准语言概念介绍SQL语言分类SQL基本概念工业互联网平台综合应用示例：SELECT*FROM

Table；//取出表中的所有列SELECTname，ageFROMTable；//取出表中name

和age两列与此同时SELECT

语句中可以使用函数和别名，例如

WordCount

中：SELECT

word,COUNT(word)FROM

table

GROUP

BY

word;1)

SELECT：

SELECT用于从DataSet/DataStream中选择数据，用于筛选出某些列。SQL常用算子SQL基本概念工业互联网平台综合应用示例：SELECTname，ageFROMTablewherenameLIKE‘%小明%’；SELECT*FROM

TableWHEREage=20；WHERE

是从原数据中进行过滤，那么在WHERE

条件中，Flink

SQL

同样支持=

、<、>、<>、>=、<=，以及AND

、OR

等表达式的组合，最终满足过滤条件的数据会被选择出来。并且WHERE

可以结合IN

、NOT

IN

联合使用。举个例子：SELECT

name,ageFROMTableWHERE

name

IN(SELECTnameFROMTable2)2)

WHERE：WHERE用于从数据集/流中过滤数据，与SELECT一起使用，用于根据某些条件对

关系做水平分割

，即选择符合条件的记录。SQL常用算子SQL基本概念工业互联网平台综合应用示例：SELECT

DISTINCT

name

FROM

Table;对于流式查询，计算查询结果所需的State

可能会无限增长，用户需要自己控制查询的状态范围，以防止状态过大。3)

DISTINCT：

DISTINCT

用于从数据集/流中根据

SELECT

的结果进行去重。SQL常用算子SQL基本概念工业互联网平台综合应用示例：SELECT

name,SUM(score)as

TotalScore

FROM

Table

GROUP

BYname;4)

GROUP

BY：

GROUP

BY

是对数据进行分组操作。例如：计算成绩明细表中每个学生的总分。5)

UNION

和

UNION

ALL：

UNION

用于将两个结果集合并起来

，要求两个结果集字段完全一

致

，包括字段类型、字段顺序。不同于UNION

ALL的是，UNION

会对结果数据去重。示例：SELECT*FROM

T1UNION(ALL)SELECT*FROM

T2；SQL常用算子SQL基本概念工业互联网平台综合应用示例：JOIN(将订单表数据和商品表进行关联)SELECT*FROM

Orders

INNER

JOIN

Product

ON

Orders.productId=Product.idLEFT

JOIN

与JOIN

的区别是当右表没有与左边相JOIN

的数据时候，右边对应的字段补NULL

输出，RIGHT

JOIN

相当于LEFT

JOIN

左右两个表交互一下位置。FULL

JOIN

相当于RIGHT

JOIN

和LEFTJOIN之后进行UNION

ALL

操作。示例：SELECT*FROM

Orders

LEFT

JOIN

Product

ON

Orders.productId=Product.idSELECT*FROM

Orders

RIGHT

JOIN

Product

ON

Orders.productId=Product.idSELECT*FROM

Orders

FULL

OUTER

JOIN

Product

ON

Orders.productId=Product.id6)

JOIN

：JOIN

用于把来自两个表的数据联合起来形成结果表。SQL常用算子SQL基本概念工业互联网平台综合应用◼SQL语句可以单行或多行书写，以分号结尾。◼SQL语句不区分大小写，建议关键字使用大写。◼可使用空格和缩进增加语句的可读性。◼注释。单行注释：--注释内容或#注释内容(mysql特有)；多行注释：/*注释*/。SQL通用语法SQL基本概念工业互联网平台综合应用外部数据源◼外部输入的业务数据表、输出到外部数据库的计算结果◼是用户自定义的数据库类型，平台支持对外

部数据源的增加、删除、修改、查看操作◼从数据平台获取数据连接◼内部数据源列表显示数据计算服务支持的数

据库信息，包括数据源名称、数据库类型、版本和IP地址数据源管理：包括内部数据源、外部数据源内部数据源数据源管理工业互联网平台综合应用任务流中使用的外部数据表数据，需要先在数据源管理中添加外部数据源添加外部数据源①选择外部数据源页签

，单击添加按钮数据源管理工业互联网平台综合应用数据源管理②选择需要添加的数据库类型可以选择MySQL、PostgreSQL、SQLSever、Oracle关系型数据库。

也可以选择Kadka消息中间件。工业互联网平台综合应用数据源管理③填写参数，完成添加数据源。以添加MySQL数据库为例，根据已有数据库的信息输入数据源名称、数据库版本、主机、端口等信息后进行测试连通性，测试成功后点击“添加”，完成数据源的添加。工业互联网平台综合应用数据源名称自定义数据源名称。支持连接的数据库版

本MySQL：5.7.x,8.0.xPostreSQL：12.xOracle：9.x.x,10.x.x,11.x.x,12.x.xKafka：1.1.x,2.0.x,2.1.x,2.2.x,2.3.x,2.4.x,2.5.x使用主机地址主机名或IP+

端口号。数据库名/SID输入源数据库名。使用JDBC格式jdbc:mysql://host:port/database?useUnicode=yes&characterEncoding=utf8&useSSL=false&serverTimezone=UTC用户名输入用户名。密码输入密码。说明：数据库版本支持下拉列表选择。

输入数据库名或SID时，确保与

源数据库名一致；使用JDBC时，请按照对应的JDBC格式连接数

据库，否则可能连接失败。创建数据源参数说明数据源管理工业互联网平台综合应用参数说明Kafka版本支持的Kafka版本有：1.1.x、2.0.x、2.1.x、2.2.x、2.3.x、2.4.x、2.5.x集群地址输入集群地址IP及端口号，如需添加多个集群地址，使用“,”区分。已经添加的外部数据源，可以通过列表中的设置，进入数据源详情页面，修改数据源的基本信息。也可以通过列表右侧的图标移除不需要的数据源。编辑和删除数据源添加消息中间件数据源管理自定义数据源名称数据源名称工业互联网平台综合应用实时数据开发过程◼实时数据开发支持ApacheFlink流处理执行模式◼将设备接入平台并上报数据后，对采集的数据进行计算，或直接透传输出到指定数据库中，是实现设备运行情况分析、指标预警、故障检测、数据大屏展示等功能的重要前提◼同时，实时数据开发提供秒级返回业务指标和快速进行多维分析等实时数据化能力实时数据开发工业互联网平台综合应用实施数据开发名称说明计算任务计算任务是数据计算服务的基本计算单元，数据获取、数据处理和计算结果输出都是通过任务完成的。任务的执行过程是个DAG

有向无环图，图中的点是执行阶

段，各个执行阶段的依赖关系是图的边。数据计算会依照图中的节点和边依赖关系执行各个阶段。任务流大数据的计算、分析和处理，一般由多个任务单元组成(Hive

、Sparksql

、

Spark

、Shell等)，每个任务单元完成特定的数据处理逻辑。任务流预估出每个任

务处理所需时间，根据先后顺序，计算出每个任务的执行的起止时间，通过定时执行任务的方式，让整个系统保持稳定的运行。任务流实例每一个离线任务根据调度周期，在每一次运行时生成一个任务流实例。任务实例任务流由多个任务组成，一个任务流实例执行的过程中，每个组成该任务流的任务均生成一个任务实例，所有任务实例执行完毕，即任务流执行完成。任务实例状态包括：启动中、运行中、运行失败、停止中、已停止。在建立实时数据开发过程中，通过拖拽任务节点，建立任务节点的关联，通过可视化

DAG

形成的数据处理任务定义。

重要概念的说明表实时数据开发工业互联网平台综合应用实施数据开发名称说明任务模板数据计算服务把使用频率高的计算任务当模板存储，减少你创建任务和调试的时间。数据源计算任务获取数据的源地址和计算结果输出的目标地址。数据平台支持内部数据源，非数据平台手动添加的数据源称为外部数据源。内部数据源是数据平台内部获取数据连接。外部数据源是用于交付项目中指定的外部数据源，外部数据源可能是外部输入的业务数据表，也可能是计算结果输出到外部数据库中。输入节点计算任务数据输入的源头。支持数据平台的Kafka

订阅数据，按物模型和设备列表选择。处理节点计算任务过程数据过滤和时间聚合计算相关的处理。支持按条件筛选和Flink

SQL查询语句。输出节点计算任务结果数据的输出目的地，支持MySQL和PostgreSQL

等。

重要概念的说明续表实时数据开发工业互联网平台综合应用实时数据开发界面可以分三块区域：左边为任务流列表

，中间是节点库

，右边是节点编排界面及数据流程，单击具体某个节点会展示当前选中节点属性编辑区域。数据流启动要求实时数据开发工业互联网平台综合应用①有且仅有一个输入节点；②每个节点有且仅有一个前节点③有至少一个输出节点；④平台支持对实时数据开发任务流的创建、保存、导入、导出、删除、编辑、查询、撤回和复制等操作。数据开发的数据流需要满足以下要求才能正常启动：数据流启动要求实时数据开发工业互联网平台综合应用文件夹◼文件夹的层级最多为5级◼支持文件夹的新增、修改(编辑文件夹名称)、删除◼支持文件夹下的任务流的新增、修改、删除◼文件夹目录方便管理创建的多个任务流，例如：同时撤回和删除多个任务流任务流的操作实时数据开发工业互联网平台综合应用③单击确定完成创建②新建空白画布，在弹出框中填入自定义的名称和选择上级文件夹①单击按钮创建任务流。实时数据开发创建任务流工业互联网平台综合应用①处是节点库

②是进行任务流保存发布等操

作的区域

③是画布。可以将节点库中的节点拖入到画布进行编辑，

在编辑完成后可以在②处选择保

存、发布导入导出等操作。④在画布中以拖拽的方式进行实时数据开发。实时数据开发工业互联网平台综合应用导入导出任务流导入导出功能方便在不同的环境和组织之间复用相同功能的任务流。已发布的任务支持导出，不支持导入。查看任务流查询已创建的实时数据开发列表，可以按发布状态进行检索，也可通过任务名称进行模糊搜索。编辑任务流可以修改任务名称及描述。编辑操作只能对未发布的任务操作，已发布的任务无法编辑复制任务流可以对任务流进行复制操作，复制的任务名称为原任务名称后添加副本及随机字符作为后缀。已发布的任务如需修改，建议先复制再修改。任务流的操作实时数据开发工业互联网平台综合应用删除任务流对于未发布的任务，可进行删除操作，弹出弹窗需确定删除，删除后无法恢复。删除操作只能对未发布的任务操作，已发布的任务无法删除。任务流的操作实时数据开发工业互联网平台综合应用当任务流在发布状态时可以进行创建任务流、导出和复制操作，但不能进行编辑名称和删除操作。实时数据开发工业互联网平台综合应用输入节点处理节点输出节点实时数据物实例筛选MySQL输出多组织实时数据数据筛选PostgreSQL输出维表关联Oracle输出Flink

SQL查询Kafka输出说明：输入节点最多只能有1个，叶子节点必须是输出节点。节点类型实时数据开发工业互联网平台综合应用实时数据开发过程中的基本的节点操作如表所示。操作名称具体操作创建节点选择节点库中的一个节点类型拖拽到画布中。复制节点鼠标悬浮于需要复制的节点，点击右上角出现的第一个按钮。删除节点/连线鼠标悬浮于需要删除的节点，点击右上角出现的第二个按钮，或者选中需要删除的节点或是连线，按一下键盘中的Delete键。连接节点点击一个节点下方的连接点拖动连线至另一个节点。移动画布在画布空白处按下鼠标并移动即可拖动画布，使用鼠标滚轮可缩放画布。节点基本操作实时数据开发工业互联网平台综合应用②多组织实时数据多组织实时数据节点已经过滤了组织和模型数据，因此，多组织实时数据节点后建议使用除物实例筛选节点以外的处理节点输入节点该节点定义了需要输入和处理什么数据。每个数据处理流程只能有一个输入节点。①实时数据实时数据来源为数据平台kafka的实时数据，实时数据节点只能连接物实例筛选节点常用节点说明实时数据开发工业互联网平台综合应用创建实时数据开发需设置节点名称、拉取数据位置、数据源类型实时数据开发工业互联网平台综合应用处理节点①物实例筛选：⚫物实例筛选节点只能连接实时数据输入节点⚫节点名称：支持字符串数据，最多可输入64个字符⚫物实例：接入与建模服务中已存在的且需要使用任务流完成数据计算的所有物理设备，包括设备、复合物、网关、物应用接口等。当物实例选择多个选项时，输出映射取字段的交集即schema取交集实时数据开发工业互联网平台综合应用处理节点①物实例筛选：⚫输出映射(模型公共属性)接入与建模服

务中设备已经定义的所有属性均可映射至

数据计算服务中，根据任务流需求选择需要参加数据计算的物模型公共属性实时数据开发工业互联网平台综合应用处理节点①物实例筛选：⚫输出映射(模型非公共属性)：接入与建模服务中设备已经定义的所有属性均可映射至数据计算服务中，根据任务流需求选择需要参加数据计算的物模型非公共属性。实时数据开发工业互联网平台综合应用支持的数据类型支持的筛选条件BOOLEAN(布尔型true/false)=、!=、null、非null、在列表中、不在列表中INTEGER(64位整型)=、!=、<=、>=、<、>、在列表中、不在列表中、null、非null、在范围中、不在范围中NUMBER(64位浮点型)=、!=、<=、>=、<、>、在列表中、不在列表中、null、非null、在范围中、不在范围中STRING(64位字符型)=、!=、在列表中、不在列表中、null、非null、包含、不包含、以开始、不以开始、以结束、

不以结束、空、非空TIMESTAMP(64位时间戳)=、!=、<=、>=、<、>、null、非null、在范围中、不在范围中数据筛选选择输入节点中的某一输出字段，并按照指定条件进行筛选，只有满足条件的数据才能继续处理实时数据开发工业互联网平台综合应用◼SQL编辑框格式化：在SQL编辑框中编写完成的SQL语句，可通过格式化将原有语句的格式规范化，便于排查错误。◼SQL语句校验：校验语法，可验证已输入的SQL语句是否符合Flink

SQL语法，

如不符合页面上方会出现警告。◼窗口最大化：编辑栏窗口可放大缩小，便于查看编辑Flink

SQL查询对流入的数据，按照编写的语句进行处理，计算结果流入下一个节点实时数据开发工业互联网平台综合应用◼数据延迟时间：如果没有时间窗口的设置，实时数据产生马上就写入mysql，mysql无法抗住写入的压力可能会出现宕机。为解决此类问题，可设置数据延迟时间，解决数据丢失的问题。◼输出字段：手动指定数据类型实时数据开发工业互联网平台综合应用◼支持的时间窗口类型数据统计逻辑事件时间EventTime和触发计算的处理时间ProcessingTime◼时间窗口大小当前时间窗口建议使用分钟级，大时间窗口建议使用批处理注意：

Number

数据类型需要指定小数后几位。确保

Flink

SQL

运算准确。避免两个字

段相乘导致的小数位溢出

，

Flink

SQL

输出就为

null

会影响数据准确性实时数据开发工业互联网平台综合应用描述字段ID数据类型时间戳timestampTIMESTAMP设备IDdeviceIdSTRING物模型deviceTypeIdSTRING物标识assetIdSTRING上云时间cloud_timeTIMESTAMP租户IDtenantIdSTRING工作状态workstatusINTEGER在线状态onlineJson当前位置locationJson说明：

Flink

SQL查询节点，同步字段功能同步时，忽略Flink

SQL别名输出数据类型默认为

DECIMAL

类型，可根据需要修改为

STRING

和

INTEGER

等类型在实时数据开发中，常需要获取以下表中系统变量进行FlinkSQL查询实时数据开发工业互联网平台综合应用参数说明节点名称自定义节点的名称，例如：每天用电量。目标数据源选择在数据源管理中添加的“外部数据源”作为需要存入输出数据的数据库。目标数据表选择外部数据源中需要存放输出数据的数据库表。插入方式-插入：直接写入数据，若写入失败，则丢弃数据。-更新插入：先查询是否有该字段，没有就插入，有就按指定的方式对字段的值进行更新。插入键默认为主键PRIMARY。输出映射将输出字段和数据库字段进行匹配。选择“更新插入”还需要设置更新数据的方式，当前支持的更新策略如下：替换：将原数据替换成经过数据开发后得到的结果数据后输出。累加：将经过数据开发后得到的结果数据与原数据进行累加后输出。最大值：将经过数据开发后得到的结果数据与原数据进行对比，取较大值作为输出数据。最小值：将经过数据开发后得到的结果数据与原数据进行对比，取较小值作为输出数据。各输出字段类型支持的更新策略如下：Boolean支持无、替换。Integer、Number支持全部策略。String支持无、替换。Timestamp支持无、替换。按时间聚合输出默认勾选，勾选后，将按照下列参数设置的时间规则输出数据。若不勾选，则直接输出计算数据。时区不同经纬度地区的地方时间。选择时区后，输出到数据库中的时间为选择的时区时间。时间聚合字段选择用户自己的库表中，与时间对应的字段，例如：time。聚合时长每半小时、时、日、周、月和年输出一次数据。开始时间业务核算的时间。例如聚合时长为1分钟，开始时间为1月1日0时0分，则将在今年1月1日0时1分输出第一条数据。输出节点

：该节点定义了数据最后输出的字段、输出到什么数据库中、是否以时间聚合进行输出等实时数据开发工业互联网平台综合应用数据计算服务TIMESTAMPBOOLEANINTEGERINTSTRINGNUMBERMySQLDATATIMEDATETIMETIMESTAMPTINYINTTINYINTSMALLINTMEDIUMINTINTBIGINTFLOATDOUBLEDECIMALTINYINTSMALLINTMEDIUMINTINTBIGINTFLOATDOUBLEDECIMALCHARVARCHARTINYTEXTTEXTMEDIUMTEXTLONGTEXTDATETIMEDATETIMETINYINTSMALLINTMEDIUMINTINTBIGINTFLOATDOUBLEDECIMALMySQL输出节点◼目标数据源：指定数据库写入经过数据处理后得到的结果，支持内部数据源和外部数据源◼插入方式：插入表示直接追加数据库记录；更新插入表示基于数据库的唯一键进行更新◼输出映射：选择需要写入目标数据源的字段，根据数据类型的一致性做了筛选的过滤实时数据开发工业互联网平台综合应用更新策略说明累加将经过数据开发后得到的结果数据与原数据进行累加后输出替换将原数据替换成经过数据开发后得到的结果数据后输出最大值将经过数据开发后得到的结果数据与原数据进行对比，取较大值作为输出数据最小值将经过数据开发后得到的结果数据与原数据进行对比，取较小值作为输出数据◼时间聚合输出：MySQL写入时会按照所选的时间聚合字段和聚合时长按照数据更新的策略进行更新数据◼更新策略：数值类型的更新策略有累加、替换、最大值、最小值；字符串类型的数据有替换策略实时数据开发工业互联网平台综合应用DAG图计算校验内容：◼流计算每个节点只有唯一父节点◼批计算Flink

SQL节点可以有多个父节点◼有向不循环DAG全称DirectedAcyclic

Graph，有向无环图DAG图编辑工业互联网平台综合应用◼流计算通过深度优先搜索(DFS)判断是否有节点在图中遍历两次，以判断有向图中是否存在闭环通过深度优先搜索(DFS)判断是否有节点在图中从未被遍历，以判断有向图中是否存孤立节点◼批计算通过Topological

ordering是否存在判断有向图中是否存在闭环通过深度优先搜索(DFS)判断是否有节点在图中从未被遍历，以判断无向图中是否存孤立节点DAG图编辑工业互联网平台综合应用任务流MySQL节点校验规则表节点节点枚举前端校验发布校验流任务MySQL输出节点MYSQL_OUTPUT_NODE-检查是否选择了相同的映射字段-检查是否将所有非空字段映射-检查是否选择了插入键-检查是否选择了数据源-检查是否选择了表-检查是否有勾选了输出但是没选映射字段的-检查是否输出了所有非自增的插入键字段(插入更新的情

况下)-检查是否在开启时间聚合的情况下选择了时间聚合字段、

检查时间聚合字段是否在输出字段中被映射了insert：非空字段是否全部映射(包

含了主键)upsert:-除自增insertkey外是否全部

映射-insertkey不能为空切不能全

为自增-非空主键是否全部映射DAG图编辑工业互联网平台综合应用约束条件校验完整的DAG图需满足必须有且只有一个输入节点，输入节点是Kafka或多组织实时数据节点，所有叶子节点必须是输出节点校验规则①DAG图校验；②节点配置校验最大深度图从输入节点到输出节点的最大深度为10

(MAX_DEPTH),每个节点的最大子节点个数为MAX_WIDTH,全部节点个数不大于50(MAX_TOTAL)，用于限制计算复杂度。DAG图编辑工业互联网平台综合应用以时间驱动，根据时间对数据流进行分组例如：每30秒钟

滚动时间窗口(Tumbling

Time

Window，无重叠) 滑动时间窗口(Sliding

Time

Window，有重叠)

会话窗口(Session

Window，以会话间隔驱动)以事件数量驱动例如：每一百个事件

滚动计数窗口(TumblingCount

Window，无重叠)

滑动计数窗口(SlidingCountWindow，有重叠)窗口提供了这样一种依据，决定了数据何时开始处理,处理哪些数据Count

WindowTime

Window相关概念Flink窗口工业互联网平台综合应用

采用会话持续时长作为窗口处理依据

会话窗口根据会话间隔切分不同的窗口

窗口的长度固定，以一个步长(Slide)不断向前滑动

步长设置的大小决定了Flink以多大的频率来创建新的窗口

滚动窗口的窗口长度是固定的，窗口之间不重叠。滑动窗口(Sliding

Window)滚动窗口(TumGbling

indow)会话窗口(Session

Window)Flink有3个内置Time

Window类型Flink窗口工业互联网平台综合应用

优势：Flink的时间系统中最简单的概念，不需要流和机器之间的协调。它提供了最佳的性能和最低的延迟。

劣势：前面发生数据积压，时间窗口内数据量暴增，占用内存高

优势：一般由数据生产方自身携带，从消息的产生就诞生了，不会改变。

劣势：设备离线后，没有数据上传，窗口不结算；设备上线后，窗口恢复计算，离线前统计数据延迟输出。

优势：可以防止Flink内部处理数据是发生乱序的情况。

劣势：数据进入Flink前出现数据积压或者断线缓冲数据导致数据迟到，数据统计到下个周期中。设备离线同理。事件时间(Event

Time)处理时间(Processing

Time)进入(摄取)时间

(Ingestion

Time)三种时间窗口的区别Flink窗口工业互联网平台综合应用窗口函数描述举例TUMBLE(time_attr,interval)翻滚时间窗口，将行分配给具有固定持续时间(interval)的非重叠、连续窗口。一个5分钟的滚动窗口即每五分钟分组一次，这五分钟内的数据为一组HOP(time_attr,interval,

interval)跳跃时间窗口，具有固定的持续时间(第二个interval参数)，并按指定的跳跃间隔(第一个interval参数)跳跃。一个5分钟的跳跃窗口，其中跳跃间隔为3分钟，即每五分钟分组一次，这五分钟内的数据为一组，等待三分钟再进行下一次分组SESSION(time_attr,interval)会话时间窗口没有固定的持续时间，但它们的界限由interval不活动的时间定义，即，如果在定义的间隙期间没有事件出现，则会话窗口关闭。设置interval为5分钟，若窗口在上一次发生事件至今超过5分钟，则该窗口关闭FlinkSQL-窗口函数Flink窗口工业互联网平台综合应用滚动窗口滑动窗口会话窗口SELECT

count(event),TUMBLE_START(time_attr,

INTERVAL‘30’SECOND)AS

ts_start,FROM

tumble_streamGROUP

BY

TUMBLE(time_attr,INTERVAL‘30’SECOND)SELECT

count(event),HOP_START(time_attr,INTERVAL‘30’SECOND,INTERVAL‘1’MINUTE)

ASts_start,FROM

slide_streamGROUP

BY

HOP(time_attr,INTERVAL‘30’SECOND,INTERVAL‘1’MINUTE)SELECT

count(event),SESSION_START(time_attr,INTERVAL‘30’SECOND)AS

ts_start,FROM

session_streamGROUP

BY

SESSION(time_attr,

INTERVAL‘30’SECOND)窗口函数示例Flink窗口工业互联网平台综合应用窗口使用的辅助函数辅助函数描述TUMBLE_START(time_attr,interval)HOP_START(time_attr,interval,interval)SESSION_START(time_attr,interval)返回相应翻滚、跳跃或会话窗口的包含下限的时间戳。TUMBLE_END(time_attr,interval)HOP_END(time_attr,interval,interval)SESSION_END(time_attr,interval)返回对应的翻滚、跳跃或会话窗口的独占上限的时间戳。注意：在后续的基于时间的操作中，例如区间连接和组窗口或跨窗口聚合中，不能将独占上限时间戳用作行时间属性。TUMBLE_ROWTIME(time_attr,interval)HOP_ROWTIME(time_attr,interval,interval)SESSION_ROWTIME(time_attr,interval)返回相应翻滚、跳跃或会话窗口的包含上限的时间戳。

结果属性是一个行时间属性，可用于后续基于时间的操作，例如间隔连接和组窗口或窗口聚合。TUMBLE_PROCTIME(time_attr,interval)HOP_PROCTIME(time_attr,interval,interval)SESSION_PROCTIME(time_attr,interval)返回一个proctime属性，该属性可用于后续基于时间的操作，例如间隔连接和组窗口或窗口聚合。TUMBLE_START(time_attr,interval)HOP_START(time_attr,interval,interval)SESSION_START(time_attr,interval)返回相应翻滚、跳跃或会话窗口的包含下限的时间戳。Flink窗口工业互联网平台综合应用《工业互联网平台综合应用》/知识准备任务实施目录0102数据源分为外部数据源和内部数据源，外部数据源支持MySQL

，

PostgreSQL、Oracle。

可以根据实践应用场景进行添加。本次任务采用外部数据源，建立外部数据源的步骤为：1)登录平台，进入控制台。2)数据计算-

>数据源管理。3)选择外部数据源，单击“添加”。连接外部数据源工业互联网平台综合应用在外部数据源可以选择MySQL、PstagreSQL、SQLserver、Oracle关系型数据库，还可以添

加消息中间件Kadka。如图所示，本次任务选择关系型MySQL数据库。连接外部数据源工业互联网平台综合应用在选择数据库类型后，根据已有数据库的信息输入数据源名称、数据库版本、主机、端口等信息后进行测试连通性，测试成功后点击“添加”，完成数据源的添加。数据源管理工业互联网平台综合应用(1)添加文件夹1)创建一级文件夹，操作步骤：登录控制台>数据计算>数据开发>实时数据开发>点击符号>添加一级文件夹，如图所示：创建实时数据开发工业互联网平台综合应用2)自定义文件夹名称，点击“确定”，完成文件夹创建。创建实时数据开发工业互联网平台综合应用(2)创建任务流1)如图所示，创建任务流。创建实时数据开发工业互联网平台综合应用2)创建实时数据开发，自定义任务名称“机器人产量分析”，上级文件夹选择“课程项目”，点击“确定”，完成实时数据开发创建。创建实时数据开发工业互联网平台综合应用1)选择输入节点“实时数据”，将节点拖拽至画布中。2)编辑输入节点，自定义节点名称，拉取位置选择“从上一次的位置拉取数据”，数据类型选择“实时数据”。输入节点设置工业互联网平台综合应用1)选择处理节点“物实例筛选”，将节点拖拽至画布中，然后点击“实时数据”节点下方的连接点拖动连线至“物实例筛选”节点。。2)自定义节点名称，选择前面建立的物实例“机器人”，在输出映射(模型公共属性)处选择导入、导出和编辑属性，完成输出映射

设置后点击“保存”。物实例筛选工业互联网平台综合应用1)采用Flink

SQL查询对流入的数据进行

处理查询，将处理节点“Flink

SQL查询”拖

拽到画布中，然后点击“物实例筛选”节点下方的连接点拖动连线至“Flink

SQL查询”节点。2)根据业务需求编写的语句进行处理，计算结果流入下一个节点，以查询每日产量为例，如下图所示，自定义节点名称为“每日产量”。FlinkSQL查询节点工业互联网平台综合应用SELECTtumble_start(rowtime,

interval

'1'

day)

as

__timestamp__,__deviceId__,LAST_VALUE(today_output)

AS

output,DATE_FORMAT(CURRENT_TIMESTAMP,'yyyyMMdd')

as

record_dateFROM{}group

bytumble(rowtime,

interval

'1'

day),__deviceId__3)编写Flink

SQL查询语句，本次任务以“查询每日产量数据”为例进行编写。其查询逻辑如下：FlinkSQL查询节点工业互联网平台综合应用4)SQL编辑框格式化：在SQL编辑框中编写完成的SQL语句，可通过格式化将原有语句的格式规范化，便于排查错误SQL语句校验：校验语法，可验证已输入的SQL语句是否符合FlinkSQL语法窗口最大化：编辑栏窗口右上角符号可控制窗口的缩放，便于查看编辑。FlinkSQL查询节点工业互联网平台综合应用5)数据延迟时间：为了缓解写“MySQL”节点的压力，平台设置了数据延误时间。如果没有时间窗口的设置，实时数据产生马上就写入mysql，mysql无法抗住写入的压力可能会出现宕机。为解决此类问题，可设置数据延迟时间，解决数据丢失的问题。设置1分钟时延。FlinkSQL查询节点工业互联网平台综合应用6)输出字段设置。如图所示，首先点击

“同步字段”，然后根据实际需求手动调整指

定输出数据类型，将“__deviced__”字段的输出数据类型设置为“String”，将“output”

字段的数据类型设置为“Number”，将

“record_date”字段的数据类型设置为“String”。最后点击“保存”，完成FlinkSQL节点的设置。FlinkSQL查询节点工业互联网平台综合应用1)将处理节点“MySQL”拖拽到画布中，然后点击

“Flink

SQL查询”节点下方的连接点拖动连线至

“MySQL”节点2)点击“MySQL”输出节点，自定义输出节点名称。3)目标数据源，选择自己所创建的外部数据源。4)目标数据表，选择自己所创建的目标数据表。5)插入方式选择“插入”6)选择输出字段，在数据库字段选择与输出字段对应的在数据库中目标数据表的字段输出节点设置工业互联网平台综合应用1)点击保存图标进行保存，然后再点击飞机图案按钮，发布实时数据开发。发布实时数据工业互联网平台综合应用3)任务发布成功，弹出“数据处理任务发布成功”弹窗。此时，实时数据开发的状态图标由灰色变成绿色。2)页面弹出弹框“确定要发布该任务吗？”，点击确定。发布实时数据工业互联网平台综合应用Thanks

工业互联网平台实时数据开发《工业互联网平台综合应用》/2

工业设备任务流运维《工业互联网平台综合应用》/知识准备任务实施目录0102实时任务运维为了方便运维工程师排查任务流执行问题，根云平台提供了数据运维功能。运维工程师能够查看

在数据计算中已经完成数据开发且已经启动任务的执行状态。数据运维是对任务的统一管理界面，可以查看任务ID、名称、运行状态、开始时间和状态更新时间等任务信息，也可以对任务进行启动、停止、撤回、查看任务详情等操作。运维工程师能够对运行状态及实时任务名称进行搜索工业互联网平台综合应用①停止：任务发布即生成运行的实时任务，可停止实时任务。②启动：运行状态为已停止的实时任务可重新启动。③撤回：对实时任务执行撤回操作后，该任务回到未发布的实时数据开发状态。④查看任务：查看实时任务的开发情况。实时任务运维对实时任务的操作：工业互联网平台综合应用运行状态图标初始化中运行中停止中运行失败实时任务运维实时任务运维的任务流运行状态有4种状态：初始化中、运行中、停止中、运行失败。工业互联网平台综合应用实时任务运维运行详情：对于运行中的实时任务

，可查看运行详情

，包括运行概况和异常日志。运行详情，当实时任务运行正常但出现数据丢失或数据库中不存在数据时，则可以使用运行详情对数据的输入输出进行排查

，确认出现问题的具体节点。工业互联网平台综合应用实时任务运维实时任务运维详情中的运行概况能实时看到每个节点的数据接收、

发出以及运行时长等基本

信息

，方便判断和排查问题。工业互联网平台综合应用实时任务运维如果是任务流上线、停止、撤回的异常信息

，可以在异常日志中查看。工业互联网平台综合应用实时任务运维批量撤回，可以在实时任务运维中批量撤回正在运行的实时计算任务。工业互联网平台综合应用数据计算MySQLPostgreSQLOracleBOOLEAN(布尔型true/false)TINYINTBOOLBOOLINTEGER(64位整型)TINYINTSMALLINTMEDIUMINTINTBIGINTFLOATDOUBLEDECIMALBIGINTSERIALINT2INT4INT8NUMBER在任务开发过程中需要对字段进行数据类型的映射或选择，例如输入节点系统会自动匹配映射类型、FlinkSQL查询节点输出字段需选择输出数据类型。实时数据开发支持以下数据类型支持的数据类型工业互联网平台综合应用TINYINTSMALLINTMEDIUMINTINTBIGINTFLOATDOUBLEDECIMALCHARVARCHARTINYTEXTTEXTMEDIUMTEXTLONGTEXTDATETIMEDATETIMEDATATIMEDATETIMETIMESTAMPFLOAT8MONEYNUMERICFLOAT4INT4INT2INT8VARCHARCHARTEXTBITINETDATETI