工业互联网平台综合应用 课件 项目三任务3.3 IT OT数据融合计算

《工业数据处理与分析》/工业互联网系列课程

项目三

生产能源管理分析任务四IT&OT数据融合计算>>目录任务说明01《工业数据处理与分析》/工业互联网系列课程知识准备02任务实施03思考与讨论04任务说明上一个任务中，数据开发工程师已经根据能源指标体系完成工业设备数据实时计算在本次任务中，数据开发工程师需要根据业务需求，搭建离线数据开发任务任务说明学习导图目录任务说明01《工业数据处理与分析》/工业互联网系列课程知识准备02任务实施03思考与讨论04批量计算概念批量计算，也称为批处理计算，是一种数据处理方式。批量计算是指对静态数据的批量处理，批量计算处理的是“固定”、“有界”的数据集。也就是说，首先会统一收集数据，并且把数据以数据表的形式存储到数据库中，再按照不同的计算逻辑，对全部的数据进行统一的批量处理，待全部数据处理完成后，才会输出最终的结果。传统的数据处理方式面临的问题：如果数据源非常大时，往往数据的移动就要花费较长时间。传统的数据处理工具往往是单机模型，面对海量数据时，数据处理的时间也是一个很大的问题。批量计算概念批量计算的主要特点：常用于处理大量数据，这些数据可以是存储在文件系统中的数据、数据库中的数据、网络中的数据等。大量数据处理通常是在一个固定时间段内，将数据按照一定的逻辑分组，进行一次性批量处理。通常适用于离线分析和历史趋势分析，这些分析往往需要大量数据支撑。通常要求数据的一致性，即每次处理的数据都是最新的数据，而不是实时的数据。通常需要在处理的时间窗口内完成所有数据的处理，处理速度是非常重要的。批量处理离线分析数据一致性处理速度批量计算概念批量计算的典型应用场景包括但不限于：数据仓库的数据同步与更新定期将来自不同源的大量数据整合、清理、转换，并加载到数据仓库中。大规模数据分析对大数据集进行复杂的分析，如模式识别、趋势预测等，这些任务往往不需要即时反馈。日志文件分析收集和分析大量日志文件，以便进行性能监控、安全分析或用户行为分析。批量计算一般应用于那些离线批量、延时较高的静态数据处理过程，适用于实时性要求不高的场景，比如定时周期执行一个任务，任务周期可以是天级别、周级别、月级别、年级别。在计算开始前已知所有输入数据，输入数据不会产生变化，一般计算量级较大，计算时间也较长。例如今天早上一点，把昨天累积的日志，计算出所需结果。一般是根据前一日的数据生成报表，虽然统计指标、报表繁多，但是对时效性不敏感。从技术操作的角度，这部分属于批量处理的操作。批量计算概念流式计算和批量计算区别类别流式计算批量计算数据处理方式流式计算是一种实时数据处理方法，它持续地、即时地处理数据流。数据以连续的流形式进入系统并被实时处理，每接收到一条新数据或数据块时，系统会立即处理。批量计算处理的是静态的数据集。数据被累积一段时间后，形成一个完整的批次，然后作为一个整体进行处理。这种处理通常发生在数据集已经完全收集后。适用场景适用于需要即时响应的场景，如实时监控、实时分析、实时决策支持系统等。适合处理历史数据集，进行大规模的数据分析、统计汇总、报告生成等，这些场景不要求即时响应。性能考量关注延迟的最小化，即数据从输入到输出的时间应尽可能短。关注吞吐量的最大化，即单位时间内处理的数据量最大。数据处理模型基于事件的处理，每个数据项或小批数据项被视为一个事件。基于批次的处理，数据分为不同的批次，每个批次包含大量数据。离线数据节点输入节点可以是外部数据，也可以是内部数据。输入节点主要包含hive，MySQL，PostgreSQL，Oracle，工业数据编织。外部数据库MySQL输入节点配置节点名称输入数据源输出映射查询语句支持字符串数据，最多可输入64个字符（1个汉字为2个字符）。节点配置可以选择MySQL外部数据源和数据源下的数据库表。节点配置选择了数据库表后，可以自动显示数据库表的字段列表，包括输入字段、输出字段名称、输出字段ID、输出数据类型。支持Where

orderby

limit

offset关键词，查询最大记录数限制1000000之内。输入节点离线数据节点处理节点FlinkSQL语法：平台使用Flink1.12.2版本，具体操作请参考“Flink1.12官方文档”或“阿里云FlinkSQL文档”SQL编辑框格式化：在SQL编辑框中编写完成的SQL语句，可通过格式化将原有语句的格式规范化，便于排查错误。SQL语句校验：校验语法，可验证已输入的SQL语句是否符合FlinkSQL语法，如不符合页面上方会出现警告。窗口最大化：编辑栏窗口可放大缩小，便于查看编辑。输出字段：手动指定输出数据类型FlinkSQL查询包含：数据筛选，FlinkSQL查询，表关联，去重。注意：Number数据类型需要指定小数后几位，确保FlinkSQL运算准确。避免如下情况出现：Number类型1和2的字段数值为小数后12位，两个字段相乘小数位就会溢出，FlinkSQL输出就为null会影响数据准确性。离线数据节点常需要获取以下内置参数进行FlinkSQL查询：内置参数字段ID数据类型说明zoonIdSTRING时区ID，表示不同的时区，例如UTC+00:00，UTC+08：00systemDatetimeSTRING任务调度时间，格式：yyyy-MM-ddHH:mm:ssbizStartDatetimeSTRING业务开始时间，格式：yyyy-MM-ddHH:mm:ssbizEndDatetimeSTRING业务结束时间，格式：yyyy-MM-ddHH:mm:ss离线数据节点输出节点目标数据源：指定数据库写入经过数据处理后得到的结果插入键：需选择数据库表的主键或者唯一索引作为更新数据的唯一标识输出映射：计算过程中Flink数据类型需要按MySQL数据类型输出，输出映射支持类型匹配，有明确支持的MySQL数据类型，以及Flink数据类型可转换成MySQL类型的映射表说明：可以查询MySQL数据表的主键字段，如果主键字段没有保存，节点配置会提示错误。。任务流编排任务流比较多的时候，如果出现问题恢复计算逻辑执行重跑和补数据操作繁琐例如：用电的小时任务执行失败，需要把小时任务、天任务和月任务都手动执行重跑，而且手动重跑的顺序必须是小时、天和月逐个完成用电的统计小时定时任务、天定时任务和月定时任务按序调度执行，只有前一个任务执行成功的情况，后一个任务才能执行，在项目实施过程需要把任务的定时时间错开；例如：8:00执行小时任务，8:15执行天任务，8:30执行月定时任务。如果时间间隔设计不合理，或者任务调度出现了延迟执行情况，就会出现天和月级别任务在小时任务未完成的情况下提前执行。任务流编排任务流编排实现了把电的小时、天和月3个任务放到一个任务流中，方便出现问题及时定位问题①在离线数据开发中创建2个离线任务编排（task1和task2）并发布；②创建任务流编排，使用task1和task2两个离线任务节点；③发布任务流编排；④配置调度每隔1小时执行；⑤在任务流运维中查看任务流的运行详情，检查运行状态。任务流编排创建编排在弹出框中填入自定义的称和选择上级文件夹。单击确定完成创建平台自动跳转到开发页面，在画布中以拖拽的方式进行任务流编排从文件夹右侧单击创建任务流编排或在开发空白页单击创建任务流创建任务流编排任务流编排离线任务节点创建任务流编排，拖入一个离线任务节点，不进行任何配置后单击保存，保存成功或提示具体错误信息。任务调度拖拽式离线计算任务节点，选择已发布的离线任务，选择之后会自动填写节点名称和描述。多个离线任务节点依赖顺序执行能成功运行。前一个节点运行失败，后一个节点的运行状态也为运行失败。说明：一个离线任务节点只能有一个后续节点任务流编排流程节点流程节点：流程节点包含依赖节点和分支节点依赖节点：依赖节点支持选择某一任务流的具体节点或者全部节点，多个任务流之间支持AND和OR的逻辑分支节点：当分支节点只有一个上游节点时，逻辑判断条件为上游节点执行状态成功和失败执行不同的下游节点当分支节点存在多个上游节点时，逻辑判断条件为上游节点状态的逻辑组合AND和OR，最终的结果为成功和失败执行不同的下游节点调度计划离线数据计算通常需要大量的计算资源和存储资源。通过制定合理的调度计划，可以优化资源的利用率，避免资源的浪费和闲置。合理分配计算任务和提前预定资源，可以使离线计算作业在最短的时间内完成，从而提高整体的计算效率。调度计划配置调度计划，离线数据开发需要系统定期自动执行。基于给定的时间点、时间间隔或者执行次数自动执行任务，可以配置批处理的执行计划，隔多久执行一次计划等。1）生效日期：调度计划生效时间及截至时间，在生效日期内，任务流按照调度周期重复执行。2）数据时区：根据指定的时区进行生效日期的计算。场景举例：在中国编排的离线数据计算任务需发布于其他国家执行时，数据时区应选择任务执行地的时区。3）调度周期：离线数据开发任务可以根据调度周期重复执行，分别为按小时、按天、按周、按月。业务指标的获取调度周期时间分类细分项说明按小时每小时重复重复时间点指定在每个小时的第几分钟执行间隔间隔指定间隔几小时重复一次重复时间点指定在该小时的第几分钟执行指定时间执行小时指定在一天内的第几小时执行，可多选重复时间点指定在该小时的第几分钟执行按天时间点指定在每天中哪一个时刻执行按周执行周次指定在每周的星期几执行，可多选时间点指定在该天中哪一个时刻执行按月执行日期指定每月的哪一天执行，可多选时间点指定在该天中哪一个时刻执行调度计划4）业务开始和结束时间：每次任务执行需要获取数据的时间段。业务逻辑：根据调度时间计算当前任务运行时，需要获取哪一时间段内的数据。举例：最近一次调度计划的运行时间为2021年11月11日12：00。即本次调度计划运行时，所选取的数据为调度时间前2小时至调度时间前1小时内的数据（即2021年11月11日10：00至2021年11月11日11：00这段时间内的数据）调度计划5）调度依赖：当前任务需要依赖于其他任务执行完成才能执行时，支持编排任务依赖，可以选择多个依赖任务。选择需要依赖的前置编排任务，指定依赖周期和依赖时间。依赖周期：依赖前置任务完成的周期。依赖时间：每次任务执行前，前置任务需要提前完成的时间。举例：如图进行配置，即表示每次离线数据开发任务执行前，需保证“clh-test”编排任务在前一小时已经执行完毕。目录任务说明01《工业数据处理与分析》/工业互联网系列课程知识准备02任务实施03思考与讨论04业务指标的获取能源数据的获取从上一个任务实时数据开发中，通过数据的实时监测，可以获取到每一小时的能源数据。在数据库可以查询到数据表fee_energy_1h的属性情况字段名类型timestamptimestampdate_dayvarcharPWR_Cdoubledate_hourvarchardevice_idvarchar业务指标的获取业务指标的获取通过获取每小时的电价和能源数据，可以通过表的连接以及指标间的计算，可以得到每天每个小时单位的电费情况以及对应的能源消费。根据本任务的业务目标分析，可以得到本任务的指标点属性ID属性名称数据类型运算方案tip_energy每日尖时能源Number计算每日尖时段（19:00-22:00）的能源。peak_energy每日峰时能源Number计算每日峰时段（8:00-11:00，15:00-19:00）的能源。ordinary_energy每日平时能源Number实时计算每日平时段（23:00-0:00，00:00-7:00）的能源。valley_energy每日谷时能源Number计算每日谷时段（7:00-8:00，11:00-15:00，22:00-23:00）的能源。day_fee每日电费Number每日尖峰平谷的电费之和。month_fee每月总电费Number每日总电费之和部分属性点表创建离线数据任务添加文件夹创建文件，操作步骤为：登录控制台>数据计算>数据开发>离线数据开发>单击符号>添加统计文件创建离线数据任务添加文件夹创建文件，操作步骤为：登录控制台>数据计算>数据开发>离线数据开发>单击符号>添加统计文件自定义文件夹名称“能源离线管理”，单击“确定”创建离线数据任务创建离线数据任务流在“离线数据开发”菜单栏单击创建任务流符号>单击白色方框>填写“创建离线数据开发”中的相关信息。在“离线数据开发”菜单栏单击创建任务流符号单击“空白创建”，创建离线数据开发创建离线数据任务创建离线数据任务流在弹出的提示框中填写创建任务的相关信息，自定义任务名称为“能源离线开发”上级文件夹选择“能源离线管理”最后单击“确定”，完成离线数据开发的创建此时，创建完成的离线数据开发的画布为空白画布数据计算—每小时费用、能源数据编辑输入节点将“MySQL”输入节点拖入画布后，编辑输入节点自定义节点名称为“尖峰平谷电费表”输入数据源选择“Project3”勾选“直接选表”选择数据表“fee”在输出字段会自动生成输出字段名称、输出字段ID、输出字段数据类型，可以根据业务需求添加、删除输出字段最后单击“保存”数据计算—每小时费用、能源数据编辑输入节点第二个节点输入选择“MySQL”，自定义节点名称为“每小时能源数据”输入数据源选择“Project3”勾选“直接选表”选择在数据源目录下的数据表“fee_energy_1h”最后单击“保存数据计算—每小时费用、能源数据搭建好电度电价和每小时能源数据两个输入节点编辑输入节点数据计算—每小时费用、能源数据FlinkSQL查询建立FlinkSQL查询节点，FlinkSQL查询处理节点可以进行自定义多个输入表的联合FlinkSQL联合查询计算选择节点库中“FlinkSQL查询”节点并将其拖拽到编辑页面中，自定义名称为“hour电费”，然后单击“保存”数据计算—每小时费用、能源数据FlinkSQL查询分别单击两个“MySQL”输出节点下方的连接点拖动连线至“hour电费”节点数据计算—每小时费用、能源数据FlinkSQL查询编辑FlinkSQL查询节点单击“hour电费”节点，此时可以看到关联的两个输入节点的节点名称和节点ID。输入节点对应产生节点ID，作为查询表名用于后续的FlinkSQL语句编辑。在“FlinkSQL语句”下输入数据查询的SQL语句数据计算—每小时费用、能源数据FlinkSQL查询SELECTab.date_day,ab.date_hour,ab.PWR_C*cd.fee_kasfee_h,ab.PWR_CFROMnode18841asabLEFTJOINnode33386ascdONab.date_hour=cd.date_hourSQL查询语句在SQL编辑框中编写完成的SQL语句，可通过格式化将原有语句的格式规范化，便于排查错误。在“FlinkSQL语句”下编写SQL语句进行多表查询，需要查询每小时电费、能源数据数据计算—每小时费用、能源数据FlinkSQL查询SQL语句校验：单击编辑栏窗口右上角“校验”，进行语法校验，可验证已输入的SQL语句是否符合FlinkSQL语法，如不符合页面上方会出现警告。窗口最大化：单击编辑栏窗口右上角图标，可对编辑窗口放大缩小，便于查看编辑数据计算—每小时费用、能源数据FlinkSQL查询首先单击“同步字段”手动指定数据类型，修改date_day的数据类型为“String”，date_hour的数据类型为“String”，fee的数据类型为“Double”，PWR_C的数据类型为“Double”,也可以根据业务需求手动添加字段和删除字段，最后单击保存数据计算—每小时费用、能源数据编辑输出节点将输出节点“MySQL”拖拽到画布中，自定义输出节点名称为“输出”单击“FlinkSQL查询”节点下方的连接点拖动连线至“MySQL”节点数据计算—每小时费用、能源数据编辑输出节点选择对应的外部数据源及目标数据表数据计算—每小时费用、能源数据编辑输出节点在选择外部数据源和目标数据表后自动生成插入键“PRIMARY”和插入键字段选择输出映射字段以及对应的数据库字段并单击保存数据计算—每小时费用、能源数据发布离线数据计算单击<发布>按钮页面弹出弹框“确定要发布该任务吗？”，单击确定。任务发布成功则会弹出“数据处理任务发布成功”数据计算—每小时费用、能源数据创建任务流编排创建文件，操作步骤为：登录控制台>数据计算>数据开发>任务流编排>单击右上角符号>添加一级文件夹。在“任务流编排”菜单栏单击一级文件夹旁边的符号，创建任务流编排自定义任务名称为“能源离线开发”并选择上级文件夹，然后单击“确定”数据计算—每小时费用、能源数据创建任务流编排选择离线任务并将其拖拽到画布中选择离线任务“能源离线开发”自定义节点名称为“能源离线开发”最后单击“保存”数据计算—每小时费用、能源数据发布任务流编排对建立的任务流编排进行保存后，单击飞机图标发布任务流编排。页面弹出弹框“确定要发布该任务吗？”单击确定，任务发布成功则会弹出“数据处理任务发布成功”单击“马上配置调度”可对任务流编排进行调度配置数据计算—每小时费用、能源数据发布任务流编排设置配置调度计划相关参数，单击保存撤回任务流编排选择任务流编排界面，单击撤回图标来对任务流编排进行撤回。页面弹出弹框“您确定要撤回任务吗？”，单击确定。任务撤回成功则会弹出“数据处理任务撤回成功”。数据计算—综合业务指标计算撤回离线任务单击撤回图标来对离线任务进行撤回。页面弹出弹框“您确定要撤回任务吗？”，单击确定。如图所示，任务撤回成功则会弹出“数据处理任务撤回成功”。添加FlinkSQL查询选择节点库中“FlinkSQL查询”节点并将其拖拽到编辑页面中自定义节点名称为“每日基本信息”，单击“保存”单击名称为“hour电费”的“FlinkSQL查询”输出节点下方的连接点拖动连线至新建的“FlinkSQL查询”节点上方数据计算—综合业务指标计算添加FlinkSQL查询自定义节点名称为“每日基本信息”，由上一节点“hour电费”对应产生节点ID为“node1750”在“FlinkSQL语句”下输入数据查询的SQL语句数据计算—综合业务指标计算添加FlinkSQL查询在SQL编辑框中编写完成的SQL语句，可通过格式化将原有语句的格式规范化，便于排查错误在“FlinkSQL语句”下编写SQL语句进行多表查询，需要查询一级指标相关数据数据计算—综合业务指标计算SELECTdate_day,SUM(fee_h)fee_day,SUM(PWR_C)energy_day,CASEWHENdate_hour>=19anddate_hour<22THENsum(fee_h)ELSE0ENDastip_fee,FlinkSQL查询逻辑数据计算—综合业务指标计算CASEWHEN(date_hour>=8anddate_hour<11)or(date_hour>=15anddate_hour<19)THENsum(fee_h)ELSE0ENDaspeak_fee,CASEWHEN(date_hour>=23)or(date_hour>=0anddate_hour<7)THENsum(fee_h)ELSE0ENDasordinary_fee,CASEWHEN(date_hour>=11anddate_hour<15)or(date_hour>=22anddate_hour<23)THENsum(fee_h)ELSE0ENDasvalley_fee,CASEWHENdate_hour>=19anddate_hour<22THENsum(PWR_C)ELSE0ENDastip_energy,CASEWHEN(date_hour>=8anddate_hour<11)or(date_hour>=15anddate_hour<19)TH