草庐IT

基于flink的数字集成方案

移动Labs 2023-03-28 原文

一、概述

在实际私有化物联网平台项目中，部分存量设备由于异构总线、多制式以太网、协议多样化等因素导致无法直接连接物联网平台，大量数据较难集成，平台侧和设备侧面临大量定制化开发，成本较高。因此难以推动客户或设备厂商进行存量设备接入改造，导致设备无法直连物联网平台，无法达到物联网平台对企业所有设备数据进行统一纳管。

企业内部存量的数据采集系统多为“烟囱式”，各个厂商的系统只需对接自己厂商的设备即可，数据孤岛问题突出。

各“烟囱”的数据格式各不相同，定制化采集任务代码不可复用，费时费力，难以同时支撑多个项目。

除了设备数据采集外，还有业务数据采集需求，传统物联网系统只能采集设备数据而无法集成业务数据。

二、技术选型

数字集成技术通过对不同系统数据的抽取(Extract)，数据清洗和转换(Transformation)以及输入最终的目标系统(Load)，打通各个业务孤岛，实现数据互联互通，助力企业数字化转型。由于物联网场景下的数据处理大多都要求实时性，所以要求实现时具备实时数据处理能力。实时计算也被称作流计算，代表是Storm、Spark Streaming、Flink等大数据技术。计算引擎也在不断更新迭代，从第一代的Hadoop MapReduce，到第二代的Spark，再到第三代的Flink技术，从批处理到微批，再到真正的流式计算。

Apache Flink是一个开源的流处理框架，应用于分布式、高性能、高可用的数据流应用程序。可以处理有限数据流和无限数据，即能够处理有边界和无边界的数据流。无边界的数据流就是真正意义上的流数据，所以Flink是支持流计算的。Flink可以部署在各种集群环境，可以对各种大小规模的数据进行快速计算。

Flink框架具备强大的流式ETL的能力，依靠其丰富的算子实现。

2.1 Source算子

Flink可以使用StreamExecutionEnvironment.addSource(source)来为我们的程序添加数据来源。

Flink已经提供了若干实现好的source functions，当然也可通过实现SourceFunction来自定义非并行的source或者实现ParallelSourceFunction接口或者扩展RichParallelSourceFunction来自定义并行的source。

Flink在流处理上的source大致有4大类：

  • 基于本地集合的source(Collection-based-source)
  • 基于文件的source(File-based-source)- 读取文本文件，即符合TextInputFormat规范的文件，并将其作为字符串返回
  • 基于网络套接字的source（Socket-based-source）- 从socket读取。元素可以用分隔符切分。
  • 自定义的source（Custom-source）
使用自定义Source算子可实现丰富的数据抽取功能。

2.2 Transform转换算子

① map

将DataStream中的每一个元素转换为另外一个元素，如将元素x变为原来的5倍：

dataStream.map { x => x * 5 }

② FlatMap

采用一个数据元并生成零个，一个或多个数据元。如，将句子分割为单词的flatmap函数：

dataStream.flatMap { str => str.split("  ") }

③ Filter

计算每个数据元的布尔函数，并保存函数返回true的数据元。如，过滤掉零值的过滤器：

dataStream.filter { x != 0 }

当然flink还具备很多其他功能的转换算子，如KeyBy、Reduce、Aggregations等，通过丰富的转换算子，flink可实现对数据的清洗和转换功能。

2.3 Sink算子

Flink的sink算子支持将数据输出到：本地文件、本地集合、HDFS，除此之外，还支持：sink到kafka、sink到mysql、sink到redis以及自定义sink算子。

通过自定义sink算子将清洗转换完成的数据输入目标系统。

三、数字集成实现

实现过程如下：

第一步，抽象定义基础控件类

数字集成基于flink可抽象定义3类基础功能控件，每类控件又可根据不同的功能实现具体的子类功能控件；详细如下：

基础功能控件分为三类：数据源控件、数据输出控件、数据处理控件。

数据源控件：将Source算子抽象定义成具备抽取数据功能的数据源控件类，并制定相应的配置规范，使用时只需根据规范配置文件，系统根据配置文件创建具体的实例化对象，实现数据抽取功能；

数据操作控件：根据不同的基础功能需求将Transform算子抽象成数据处理控件类，制定相应的配置规范，使用时只需根据规范配置文件，系统根据配置创建相应的实例化对象实现数据处理功能；

数据输出控件：将Sink算子抽象成数据输出控件类，制定相应的配置规范，使用时只需根据规范配置文件，系统根据配置创建实例化对象实现数据输出功能。

同时系统内部明确定义flink算子之间流转的数据格式作为内部流转数据格式以及根据配置输出每个基础功能控件输出的数据格式。

第二步，根据抽象定义的基础功能控件，制定具体配置规范

基础功能控件规范如下：

通过以上两步规范定义后，在同一个系统中，同一个处理过程只需要定义一个基础功能控件规范。如Kafka消费者所需的配置如Kafka集群地址、消费群组、数据所在topic、数据所在分区key，消费位置等，只需要规定上述举例这样一个Kafka消费控件并开发实现，该控件就可以在该系统中复用，每次配置的数据处理工作流，复用Kafka消费控件类并根据新配置的源系统提供的Kafka集群地址、数据所在topic等配置即可实例化该工作流所需的kafka 消费者，实现过程从开发无数次Kafka Consumer的代码变为实现一次Kafka Consumer控件代码，大量节省开发时间和开发成本。

第三步，通过对基本功能的抽象，实现如HTTP请求、kafka生产、数据遍历、条件循环、数据映射、MySQL写操作等基础功能控件并实现，再根据各个基础功能运行的先后逻辑组装相应配置执行脚本来编排组建成一个完整flink流处理链路，即可完成不同系统间的数据集成功能。

如在私有化项目中有将设备厂商云平台中智能门锁状态信息同步至自有云平台进行智能门锁控制的需求，由于智能门锁设备协议与自有物联网平台数据采集协议不适配，无法直连，由设备厂商云平台提供智能门锁状态信息推送功能，由自有物联网平台提供推送数据接收接口，完成智能门锁状态信息的同步功能。

在此案例中，通过flink框架的自定义Source算子实现HTTP POST功能接口的HTTP监听控件完成设备厂商云平台的推送数据接收功能，将接收到的智能门锁状态信息根据智能门锁ID、状态status与自有云平台存储的状态进行比较的IF分支控件，将存在状态变化的智能门锁状态信息数据向后序Sink算子流转，通过自定义Sink算子实现自有云平台数据上传功能，完成智能门锁状态信息的跨平台更新功能。

第四步，根据组建好的执行逻辑生成有向无环图，提交Flink运行，具体如下：

通过对不同的基础功能控件，基于有向无环图，将基础功能控件放入有向无环图的顶点，其中整个图中只有一个数据源控件，且无其他基础功能控件可以将数据传输给它；数据输出控件和数据操作控件可以多个，对应多条分支处理逻辑。将数据传输方向作为有向无环图的边，以此连接和组织跨系统数据传输过程中针对数据的不同逻辑顺序，生成一条完整的数据传输处理链路，将此图完整实现，提交flink执行，即可实现完整的数据抽取、转换以及输出的数字集成功能。

四、总结

最后我们来总结下基于Flink的数字集成能力的实现。得益于flink在ETL数据集成上的丰富能力以及算子之间易于处理的基础功能，我们将flink的3类算子进行抽象定义实现3类基础功能控件，实现不同的数据处理过程。根据不同的功能需求，通过Source算子实现从消息队列、API、数据库等多种数据源抽取数据的功能；通过丰富的Transform算子实现数据的清洗、筛选、转换的功能；最后可通过Sink算子实现将目标格式数据输入目标系统接收数据的渠道如消息队列、数据库、API等。综上所述，基于Flink的数字集成能力是可以实现并且具备丰富功能和可扩展性的。

flink基于spanstylestrong移动开发$flink数字集成

有关基于flink的数字集成方案的更多相关文章

  1. ruby - 在 jRuby 中使用 'fork' 生成进程的替代方案？ - 2

    在MRIRuby中我可以这样做:deftransferinternal_server=self.init_serverpid=forkdointernal_server.runend#Maketheserverprocessrunindependently.Process.detach(pid)internal_client=self.init_client#Dootherstuffwithconnectingtointernal_server...internal_client.post('somedata')ensure#KillserverProcess.kill('KILL',

  2. ruby-on-rails - 如何使辅助方法在 Rails 集成测试中可用？ - 2

    我在app/helpers/sessions_helper.rb中有一个帮助程序文件，其中包含一个方法my_preference，它返回当前登录用户的首选项。我想在集成测试中访问该方法。例如，这样我就可以在测试中使用getuser_path(my_preference)。在其他帖子中，我读到这可以通过在测试文件中包含requiresessions_helper来实现，但我仍然收到错误NameError:undefinedlocalvariableormethod'my_preference'.我做错了什么？require'test_helper'require'sessions_hel

  3. ruby-on-rails - 我如何将 Hoptoad 与 DelayedJob 和 DaemonSpawn 集成？ - 2

    我一直很高兴地使用DelayedJob习惯用法:foo.send_later(:bar)这会调用DelayedJob进程中对象foo的方法bar。我一直在使用DaemonSpawn在我的服务器上启动DelayedJob进程。但是...如果foo抛出异常，Hoptoad不会捕获它。这是任何这些包中的错误...还是我需要更改某些配置...或者我是否需要在DS或DJ中插入一些异常处理来调用Hoptoad通知程序？回应下面的第一条评论。classDelayedJobWorker 最佳答案 尝试monkeypatchingDelayed::W

  4. ruby - 查找字符串中的内容类型(数字、日期、时间、字符串等) - 2

    我正在尝试解析一个CSV文件并使用SQL命令自动为其创建一个表。CSV中的第一行给出了列标题。但我需要推断每个列的类型。Ruby中是否有任何函数可以找到每个字段中内容的类型。例如，CSV行:"12012","Test","1233.22","12:21:22","10/10/2009"应该产生像这样的类型['integer','string','float','time','date']谢谢! 最佳答案 require'time'defto_something(str)if(num=Integer(str)rescueFloat(s

  5. 叮咚买菜基于 Apache Doris 统一 OLAP 引擎的应用实践 - 2

    导读：随着叮咚买菜业务的发展，不同的业务场景对数据分析提出了不同的需求，他们希望引入一款实时OLAP数据库，构建一个灵活的多维实时查询和分析的平台，统一数据的接入和查询方案，解决各业务线对数据高效实时查询和精细化运营的需求。经过调研选型，最终引入ApacheDoris作为最终的OLAP分析引擎，Doris作为核心的OLAP引擎支持复杂地分析操作、提供多维的数据视图，在叮咚买菜数十个业务场景中广泛应用。作者｜叮咚买菜资深数据工程师韩青叮咚买菜创立于2017年5月，是一家专注美好食物的创业公司。叮咚买菜专注吃的事业，为满足更多人“想吃什么”而努力，通过美好食材的供应、美好滋味的开发以及美食品牌的孵

  6. 区块链之加解密算法&数字证书 - 2

    目录一.加解密算法数字签名对称加密DES(DataEncryptionStandard)3DES(TripleDES)AES(AdvancedEncryptionStandard)RSA加密法DSA(DigitalSignatureAlgorithm)ECC(EllipticCurvesCryptography)非对称加密签名与加密过程非对称加密的应用对称加密与非对称加密的结合二.数字证书图解一.加解密算法加密简单而言就是通过一种算法将明文信息转换成密文信息，信息的的接收方能够通过密钥对密文信息进行解密获得明文信息的过程。根据加解密的密钥是否相同，算法可以分为对称加密、非对称加密、对称加密和非

  7. 基于C#实现简易绘图工具【100010177】 - 2

    C#实现简易绘图工具一.引言实验目的:通过制作窗体应用程序(C#画图软件),熟悉基本的窗体设计过程以及控件设计,事件处理等,熟悉使用C#的winform窗体进行绘图的基本步骤,对于面向对象编程有更加深刻的体会.Tutorial任务设计一个具有基本功能的画图软件**·包括简单的新建文件,保存,重新绘图等功能**·实现一些基本图形的绘制,包括铅笔和基本形状等,学习橡皮工具的创建**·设计一个合理舒适的UI界面**注明:你可能需要先了解一些关于winform窗体应用程序绘图的基本知识,以及关于GDI+类和结构的知识二.实验环境Windows系统下的visualstudio2017C#窗体应用程序三.

  8. jenkins部署1--jenkins+gitee持续集成 - 2

    前置步骤我们都操作完了，这篇开始介绍jenkins的集成。话不多说，看操作1、登录进入jenkins后会让你选择安装插件，选择第一个默认的就行。安装完成后设置账号密码，重新登录。2、配置JDK和Git都需要执行路径，所以需要先把执行路径找到，先进入服务器的docker容器，2.1JDK的路径root@69eef9ee86cf:/usr/bin#echo$JAVA_HOME/usr/local/openjdk-82.2Git的路径root@69eef9ee86cf:/#whichgit/usr/bin/git3、先配置JDK和Git。点击：ManageJenkins>>GlobalToolCon

  9. kvm虚拟机安装centos7基于ubuntu20.04系统 - 2

    需求：要创建虚拟机，就需要给他提供一个虚拟的磁盘，我们就在/opt目录下创建一个10G大小的raw格式的虚拟磁盘CentOS-7-x86_64.raw命令格式：qemu-imgcreate-f磁盘格式磁盘名称磁盘大小qemu-imgcreate-f磁盘格式-o?1.创建磁盘qemu-imgcreate-fraw/opt/CentOS-7-x86_64.raw10G执行效果#ls/opt/CentOS-7-x86_64.raw2.安装虚拟机使用virt-install命令，基于我们提供的系统镜像和虚拟磁盘来创建一个虚拟机，另外在创建虚拟机之前，提前打开vnc客户端，在创建虚拟机的时候，通过vnc

  10. ruby - 将n维数组的每个元素乘以Ruby中的数字 - 2

    在Ruby中，是否有一种简单的方法可以将n维数组中的每个元素乘以一个数字？这样:[1,2,3,4,5].multiplied_by2==[2,4,6,8,10]和[[1,2,3],[1,2,3]].multiplied_by2==[[2,4,6],[2,4,6]]？(很明显，我编写了multiplied_by函数以区别于*，它似乎连接了数组的多个副本，不幸的是这不是我需要的)。谢谢! 最佳答案 它的长格式等价物是:[1,2,3,4,5].collect{|n|n*2}其实并没有那么复杂。你总是可以使你的multiply_by方法:c

随机推荐