Apache Beam 架构原理及应用实践
总结其他的内容都可以按照门路图里面整理出来的知识点逐一去认识，学习，消化，不建议你去看书学习，最好是多看一些视频，把不懂地方反复看，学习了一节视频内容第二天一定要去复习，并总结成思维导图，形成树状知识网络布局，方便日后复习。
这里还有一份很不错的《Java底子核心总结笔记》，特意跟各人分享出来
目录：
https://i-blog.csdnimg.cn/blog_migrate/29a6dd3c470c67197b5a324b9468f427.png
部门内容截图：
https://i-blog.csdnimg.cn/blog_migrate/72c3721170d34bc6a2ea4335f110fd54.png
https://i-blog.csdnimg.cn/blog_migrate/c37345b4d1ca0dc4561e21640489b28a.png
本文已被CODING开源项目：【一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码】收录
需要这份体系化的资料的朋侪，可以点击这里获取
DAG，中文名“有向无环图”。“有向”指的是有方向，准确的说应该是同一个方向，“无环”则指够不成闭环。如果做一些去重、统计、分组等，开辟职员不用再做 Map Reduce ，Beam 已经封装提供了相应的高级操作。
▌****Apache Beam 的架构设计
我们接下来看一下 Beam 架构是怎样的：
1. Apache Beam 的总体架构
https://i-blog.csdnimg.cn/blog_migrate/9e0445d2b3300c61fd9078cb80a1bb4f.png
Apache Beam 的总体架构是这样的，上面有各种语言，编写了差别的 SDKs，Beam 通过连接这些 SDK 的数据源举行管道的逻辑操作，末了发布到大数据引擎上去执行。需要注意的是，Local 虽然是一个 runner 但是不能用于生产上，它是用于调试/开辟使用的。
2. Apache Beam 的部署流程图
https://i-blog.csdnimg.cn/blog_migrate/1d3a525ae6bb26f7d710eb299b809d21.png
让我们一起看下 Apache Beam 总体的部署流程。首先我们去构建这个 Beam jobAPI .jar 通过 job 服务器以及设置大数据执行平台，末了提交 flink 或 spark 的使命集群去执行使命。
▌****Apache Beam 的核心组件刨析
1. SDks+Pipeline+Runners （前后端分离）
https://i-blog.csdnimg.cn/blog_migrate/a82030dc1df99eadbb98d6d09cc9da64.png
如上图，前端是差别语言的 SDKs，读取数据写入管道， 末了用这些大数据引擎去运行。可以发现完整的 beam 程序由 SDks+Pipeline+Runners 构成的。
2. 什么是 SDK？
https://i-blog.csdnimg.cn/blog_migrate/173bec4f4ecfecb647af388e66d9c4ee.png
什么是 SDK，就是一个编写 beam 管道构成的一部门，一个客户端或一个类库组件也可以，末了提交到大数据运行平台上。
3. Beam 版本和 Kafka-clients 依赖环境表
https://i-blog.csdnimg.cn/blog_migrate/7e4362dfec38c5d4bbbf87f0be3d0569.png
我们以 kafka 为例，看一下 Kafka-client 对版本的依赖环境，从图中可以看出 beam 2.6.0 版本的 api 改变根本是稳定的。固然，现在用的比较多的2.4、2.5版本。吐个槽，2.6版本之前的兼容性题目，上个版本还有这个类或方法，下一个版本就没有了，兼容性不是很好。
4. SDK beam-sdks-java-io-kafka 读取源码剖析
https://i-blog.csdnimg.cn/blog_migrate/4e60f92d455ad31bc5160300cd76270e.jpeg
https://i-blog.csdnimg.cn/blog_migrate/23e0c07eaae9f3558fc92add7a993996.png
https://i-blog.csdnimg.cn/blog_migrate/2a8362754d837eb82f610f355b401082.png
https://i-blog.csdnimg.cn/blog_migrate/6389aa107d988aeedfdabb62a6cd1507.png
https://i-blog.csdnimg.cn/blog_migrate/40cec01c18e1c2eed5a2e371e95655b8.png
① 指定 KafkaIO 的模型，从源码中不丢脸出这个地方的 KafkaIO<K,V> 范例是 Long 和 String 范例，也可以换成其他范例。
pipeline.apply(KafkaIO.<Long, String>read() pipeline.apply(KafkaIO.<Long, String>read()
② 设置 Kafka 集群的集群地点。
.withBootstrapServers(“broker_1:9092,broker_2:9092”)
③ 设置 Kafka 的主题范例，源码中使用了单个主题范例，如果是多个主题范例则用 withTopics(List) 方法举行设置。设置环境根本跟 Kafka 原生是一样的。
.withTopic(“my_topic”) // use withTopics(List) to read from multiple topics.
④ 设置序列化范例。Apache Beam KafkaIO 在序列化的时候做了很大的简化，比方原生 Kafka 可能要通过 Properties 类去设置 ，还要加上很长一段 jar 包的名字。
Beam KafkaIO 的写法：
.withKeyDeserializer(LongDeserializer.class)
.withValueDeserializer(StringDeserializer.class)
原生 Kafka 的设置：
Properties props = new Properties();
props.put(“key.deserializer”,“org.apache.kafka.common.serialization.ByteArrayDeserializer”);
props.put(“value.deserializer”,“org.apache.kafka.common.serialization.ByteArrayDeserializer”);
⑤ 设置 Kafka 的消费者属性，这个地方还可以设置其他的属性。源码中是针对消费分组举行设置。
.updateConsumerProperties(ImmutableMap.of(“group.id”, my_beam_app_1"))
⑥ 设置 Kafka 吞吐量的时间戳，可以是默认的，也可以自界说。
.withLogAppendTime()
⑦ 相称于 Kafka 中 “isolation.level” , “read_committed”，指定 KafkaConsumer 只应读取非事务性消息，或从其输入主题中提交事务性消息。流处理惩罚应用程序通常在多个读取处理惩罚写入阶段处理惩罚其数据，每个阶段使用前一阶段的输出作为其输入。通过指定 read_committed 模式，我们可以在所有阶段完成一次处理惩罚。针对 “Exactly-once” 语义，支持 Kafka 0.11 版本。
.withReadCommitted()
⑧ 设置 Kafka 是否自动提交属性 “AUTO_COMMIT”，默认为自动提交，使用 Beam 的方法来设置。
set CommitOffsetsInFinalizeEnabled(boolean commitOffsetInFinalize)
.commitOffsetsInFinalize()
⑨ 设置是否返回 Kafka 的其他数据，比方 offset 信息和分区信息，不用可以去掉。
.withoutMetadata() // PCollection<KV<Long, String>>
⑩ 设置只返回 values 值，不用返回 key。比方 PCollection，而不是 PCollection<Long，String>。
.apply(Values.create()) // PCollection
在写入 Kafka 时完全一次性地提供语义，这使得应用程序可以或许在 Beam 管道中的一次性语义之上提供端到端的一次性包管。它确保写入接收器的记录仅在 Kafka 上提交一次，即使在管道执行期间重试某些处理惩罚也是如此。重试通常在应用程序重新启动时发生（如在故障规复中）或者在重新分配使命时（如在自动缩放事件中）。Flink runner 通常为流水线的结果提供准确一次的语义，但不提供变换中用户代码的副作用。如果诸如 Kafka 接收器之类的转换写入外部体系，则这些写入可能会多次发生。
在此处启用 EOS 时，接收器转换将兼容的 Beam Runners 中的查抄点语义与 Kafka 中的事务联系起来，以确保只写入一次记录。由于实现依赖于 runners checkpoint 语义，因此并非所有 runners 都兼容。Beam 中 FlinkRunner 针对 Kafka 0.11+ 版本才支持，然而 Dataflow runner 和 Spark runner 如果操作 kafkaIO 是完全支持的。
关于性能的注意事项：
“Exactly-once” 在接收初始消息的时候，除了将原来的数据举行格式化转换外，还履历了 2 个序列化 - 反序列化循环。根据序列化的数目和资源，CPU 可能会涨的很显着。通过写入二进制格式数据（即在写入 Kafka 接收器之前将数据序列化为二进制数据）可以低落 CPU 资源。
5. Pipeline
https://i-blog.csdnimg.cn/blog_migrate/987d00a3ac9232b01a1a579d971a5e15.png
[*]您输入的数据存储在那里？
首先要确定你要构造几条数据源，在 Beam 可以构建多条，构建之前可以选择本身的 SDK 的 IO。
[*]您的数据范例是什么样的？
Beam 提供的是键值对的数据范例，你的数据可能是日志文本，格式化装备事件，数据库的行，所以在 PCollection 就应该确定命据集的范例。
[*]您想怎么行止置惩罚数据？
对数据举行转换，过滤处理惩罚，窗口计算，SQL 处理惩罚等。在管道中提供了通用的 ParDo 转换类，算子计算以及 BeamSQL 等操作。
[*]您打算把数据末了输出到那里去？
在管道末尾举行 Write 操作，把数据末了写入您本身想存放或末了流向的地方。
https://i-blog.csdnimg.cn/blog_migrate/4726d8dce4e7076b444c3878270b6a49.png
告急的是要明确变换不消耗 PCollections；相反，他们会考虑 a 的每个元素 PCollection 并创建一个新 PCollection 的输出。这样，您可以对差别的元素执行差别的操作 PCollection。这里是出现了两条管，比方输入 AR，AI，VAR，BT，BMP。
https://i-blog.csdnimg.cn/blog_migrate/124059b58eca9de9ea8115949e72f2b3.png
比方差别的数据源，有数据库，文件，以及缓存等输入举行归并。
https://i-blog.csdnimg.cn/blog_migrate/5979953fa6934269cc36b959be6e2f3b.png
一种是收费的拓蓝公司出品叫 Talend Big Data Studio，有没有免费的呢？
https://i-blog.csdnimg.cn/blog_migrate/b2d06bd6bbce5f48730368399f92752e.png
有的，它叫 kettle-beam。比方差别的数据源，有数据库，文件，以及缓存等输入举行归并。各人可以去 github 去看一下插件相应的安装及使用说明。从图中可以看出大部门 beam 的输入输出现在都是支持的。
https://github.com/mattcasters/kettle-beam
6. Runners
https://i-blog.csdnimg.cn/blog_migrate/b358dacad9132335fa364185789d7362.png
我们在看一下运行平台，这是运行平台支持度的截图。比方差别的数据源，有数据库，文件，以及缓存等输入举行归并。
Runners 在 Beam Model 模型中有4个支持的维度：
[*] What，如何对数据举行计算？比方，呆板学习中练习学习模型可以用 Sum 或者 Join 等。在 Beam SDK 中由 Pipeline 中的操作符指定。
[*] Where，数据在什么范围中计算？比方，基于 Process-Time 的时间窗口、基于 Event-Time 的时间窗口、滑动窗口等等。在 Beam SDK 中由 Pipeline 的窗口指定。
[*] When，何时输出计算结果？比方，在 1 小时的 Event-Time 时间窗口中，每隔 1 分钟将当前窗口计算结果输出。在 Beam SDK 中由 Pipeline 的 Watermark 和触发器指定。
[*] How，迟到数据如那边理？比方，将迟到数据计算增量结果输出，或是将迟到数据计算结果和窗口内数据计算结果归并成全量结果输出。在 Beam SDK 中由 Accumulation 指定。
① What
https://i-blog.csdnimg.cn/blog_migrate/4486217d6e0ffcc5f14e19e30f24c0e2.png
对数据如果处理惩罚，计算。分组的矩阵图，提到这里说一下，这些运行平台已经集成到 Beam，只是没有更新到官方首页而已。以及或者是官方不打算主推的，就没有写上去。
② Where
https://i-blog.csdnimg.cn/blog_migrate/27155f2e78b2dba05ab82ddf349146cf.png
窗口处理惩罚矩阵能力图，各人从图中可以看出很多都是全部支持的。
③ When
https://i-blog.csdnimg.cn/blog_migrate/094e3e75cde4c3b99a68547285694dab.png
对于事件处理惩罚，流计算引擎Apache Flink，Google Cloud ，Dataflow 以及 Jstorm 都支持性比较好。
④ How
https://i-blog.csdnimg.cn/blog_migrate/31c95cabcaa75e2a65a02d7f754a54f7.png
末了是对迟到数据的数据处理惩罚能力矩阵图。
7. FlinkRunner Beam
https://i-blog.csdnimg.cn/blog_migrate/2d4082619b075a5452a6d4818aabfc7f.png
我们以最近两年最火的 Apache Flink 为例子，帮各人解析一下 beam 集成环境。各人可以从图中看出，flink 集成环境。
https://i-blog.csdnimg.cn/blog_migrate/61449a3b5d5eb25da287486d328dfaf6.png
然后看一下，FlinkRunner 详细解析了哪些参数，以及代码中怎样设置。
8. Beam SQL
https://i-blog.csdnimg.cn/blog_migrate/c04c6bba51062369a1a2d933c7d4cfaa.png
Apache Calcite 是一种保准 SQL 的解析器，用于大数据处理惩罚和一些流加强功能，基于它做 SQL 引擎的有很多，比方 spark，Cassandra，druid 和我们的 Beam。
https://i-blog.csdnimg.cn/blog_migrate/f87571a8a90b8da280ca371183425149.png
我们看一下 Beam SQL 的设计思路：首先是我们写的 SQL 语句，举行查询解析，验证来源的范例，数据格式，建一个执行计划，然后通过优化，设计计划规则或逻辑，封装在 Beam 管道中，举行编译器编译，末了提交 job 到运行平台执行。
https://i-blog.csdnimg.cn/blog_migrate/d92d1d8508ab745b90a74b157531bc17.png
表中是 beam SQL 和 Calcite 的范例支持度，是把 Calcite 举行映射。
https://i-blog.csdnimg.cn/blog_migrate/20ddc0d79d74e7262781c0d50e568f81.png
Beam SQL 和 Apache Calcite 函数的支持度。里面有一些现在不支持的，需要各人做的时候多多关注，特殊是架构师设计时候。
https://i-blog.csdnimg.cn/blog_migrate/ed1aa1ba202201b3bb07d3985f8e881d.png
从图中可以看出，首先要设置好数据范例，在设置数据，末了填充到管道数据集，末了做 SQL 的操作。实在这样写照旧不方便的。有没有很好的解决方式，有。各人继续往下看…
https://i-blog.csdnimg.cn/blog_migrate/b5b3e045ce822b43b84138530bfd87b1.png
Beam SQL 的扩展。Beam SQL 的 CREATE EXTERNAL TABLE 语句注册一个映射到外部存储体系的捏造表 。对于某些存储体系，CREATE EXTERNAL TABLE 在写入发生之前不会创建物理表。物理表存在后，您可以使用访问表 SELECT，JOIN 和 INSERT INTO 语句。通过捏造表，可以动态的操作数据，末了写入到数据库就可以了。这块可以做成视图抽象的。
Create 创建一个动态表，tableName 背面是列名。TYPE 是数据来源的范例，限制支持 bigquery，pubsub，kafka，text 等。Location 下面为表的数据范例配置， 这里以 kafka 为例。
▌****AloT PB 级实时数据，怎么构建本身的“AI微服务”？
在 AIoT 里面，实时性数据比较大，比方视频分析，视频发掘，合规检测，语音分析等等。130W 路的摄像头每秒写入300多 G 的视频，一天就是 25PB，有人说可以晚上用批方式上数据，实在 AIoT 场景跟其他的场景是不一样的，比方做智能儿童手表，我们晚上上报数据的频度可以变低，白天儿童上学放学路上可以正常上报数据。AIoT 场景下摄像头24小时监控的，而且宽带主杆线都换成千兆光线，实在也支持不了每秒 300G 的实时写入。我们是怎么处理惩罚呢？
https://i-blog.csdnimg.cn/blog_migrate/e6510fe17f60bedd4f27e79142379a3a.png
首先在设计架构方案的时候，信赖很多架构师都会这样想，不想第一个去吃螃蟹，因为稳定性，安全性，及不确定性缘故起因会导致整个项目的成败。那我们看一下 Beam 有哪些大厂在使用。
知道他们使用 Beam ，咱们了解一下他们用 Beam 做了什么？比方：
[*] 使用 Apache Beam 举行大规模流分析
[*] 使用 Apache Beam 运行定量分析
[*] 使用 Apache Beam 构建大数据管道
[*] 从迁徙到 Apache Beam 举行地理数据可视化
[*] 使用 Apache Beam ＆ tf.Transform 对 TensorFlow 管道举行预处理惩罚
[*] 卫星图像的土地利用分类
[*] 智慧城市大数据集成
[*] 安全城市及质量实时风控
[*] 电商平台双十一活动实时数据处理惩罚
国外的可以从官方网站上找到案例的原文，国内可以从新闻或者官方网站找到相应的案例。
在 AloT 场景下我们为什么会选择 Beam 呢？
[*] 数据源可以适配，因为安全城市，雪亮工程数据源千奇百怪。
[*] 可以或许举行数据多样处理惩罚，连接，过滤，归并，拆分。
[*] 具有洗濯脏数据功能，比方警情去重误报警，合规检测等。
[*] 具有大数据集群捏造化部署功能，可扩展性，伸缩性。
[*] 具有实时处理惩罚和离线处理惩罚能力。
1. 案列体系架构图
https://i-blog.csdnimg.cn/blog_migrate/170ddfe13d2fe79f6c3cd1dc39633624.png
这是案例的总架构图，底层是 Beam SDK，上层是抽象封装的输入输出组件，以及洗濯组件，范例管理，第三方 SDK，在往上层是组件配置管理，及版本控制，最上层是 jar 可视化配置，以及 SQL 可视化，末了把 jar 通过运维一体化平台提交给执行引擎集群，固然这里有个解析器，是我们本身开辟的。
2. 示例架构图
https://i-blog.csdnimg.cn/blog_migrate/faf908bcf3998d6965eb3da1d6e225bd.png
以下为示例架构图：
① 摄像头以及 AI 智能装备产生的报警以及抓取的信息上报到后端智能装备。
② 智能装备产生的 AI 分析结果举行通过网关集群举行传输，注意网关集群地方要做流控及雪崩控制。
③ 消息通过网关集群发送到消息中间件。注意：这边这个规则下发是针对前段的数据举行 ETL 洗濯的洗濯规则的下发。
④ Beam 集群接收下发规则的更新，而且根据规则举行数据洗濯。
⑤ 对于文档性的数据我们实时存储到实时搜刮引擎。
⑥ 需要复杂查询，统计以及报表的数据存储到 ClickHouse。
⑦ 举行 BI 套件的展示以及前端大屏幕的展示。
3. 示例代码
https://i-blog.csdnimg.cn/blog_migrate/652ab9350336a008cdcb98879b3dea3b.png
核心示例代码，首先创建管道工厂，然后显示设置执行引擎，根据 SDKIO 举行读取 kafka 的消息。
https://i-blog.csdnimg.cn/blog_migrate/e666391f60009318b7d448dae1fad31f.png
序列化消息，写入 es 举行备份，因为 es 数据是 json 的写入的时候首先要考虑转换成 json 范例。这个地方我设置了一个编码，实体类的编码范例为 AvroCoder ，编码范例是每个管道都要设置的。
末了总结
搞定算法，面试字节再不怕，有需要文章中分享的这些二叉树、链表、字符串、栈和队列等等各大面试高频知识点及解析
末了再分享一份终极手撕架构的大礼包(学习笔记)：分布式+微服务+开源框架+性能优化
https://i-blog.csdnimg.cn/blog_migrate/a1d3333b75a6a169bea4600eab90bc00.png
本文已被CODING开源项目：【一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码】收录
需要这份体系化的资料的朋侪，可以点击这里获取
Jpei5xcGljLmNuL21tYml6X3BuZy96SGJ6UVBLSUJQamxjTVh6MnFZYnlVZTJUd2JxTERWUVhtWURPUG9nSG52ZHZRNzBCdmljYll2VEVMNVhpYkdjeDRHdGliUDBPQXZPZTFrdzdkU3hJYnVKZy82NDA_d3hfZm10PXBuZw?x-oss-process=image/format,png)
序列化消息，写入 es 举行备份，因为 es 数据是 json 的写入的时候首先要考虑转换成 json 范例。这个地方我设置了一个编码，实体类的编码范例为 AvroCoder ，编码范例是每个管道都要设置的。
末了总结
搞定算法，面试字节再不怕，有需要文章中分享的这些二叉树、链表、字符串、栈和队列等等各大面试高频知识点及解析
末了再分享一份终极手撕架构的大礼包(学习笔记)：分布式+微服务+开源框架+性能优化
[外链图片转存中…(img-YylIzmLE-1715810408976)]
本文已被CODING开源项目：【一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码】收录
需要这份体系化的资料的朋侪，可以点击这里获取
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]