|
|
Flink是如何处理反压的
Flink 内部是基于 producer-consumer 模型来进行消息传递的，Flink的反压设计也是基于这个模型。Flink 使用了高效有界的分布式阻塞队列，就像 Java 通用的阻塞队列（BlockingQueue）一样。下游消费者消费变慢，上游就会受到阻塞。
如何排查生产环境中的反压问题
反压出现的场景
反压经常出现在促销、热门活动等场景。短时间内流量陡增造成数据的堆积或者消费速度变慢。
它们有一个共同的特点：数据的消费速度小于数据的生产速度。
反压监控方法
通过Flink Web UI发现反压问题。
Flink 的 TaskManager 会每隔 50ms 触发一次反压状态监测，共监测 100 次，并将计算结果反馈给 JobManager，最后由 JobManager 进行计算反压的比例，然后进行展示。
这个比例展示逻辑如下：
OK: 0 <= Ratio <= 0.10，表示状态良好正； LOW: 0.10 < Ratio <= 0.5，表示有待观察；
HIGH: 0.5 < Ratio <= 1，表示要处理了（增加并行度/subTask/检查是否有数据倾斜/增加内存）。
0.01，代表100次中有一次阻塞在内部调用。
flink反压的实现方式
Flink任务的组成由基本的“流”和“算子”构成，“流”中的数据在“算子” 间进行计算和转换时，会被放入分布式的阻塞队列中。当消费者的阻塞队列满时，则会降低生产者的数据生产速度
反压问题定位和处理
Flink会因为数据堆积和处理速度变慢导致checkpoint超时，而checkpoint是
Flink保证数据一致性的关键所在，最终会导致数据的不一致发生。
数据倾斜：可以在 Flink 的后台管理页面看到每个 Task 处理数据的大小。当数据倾斜出现时，通常是简单地使用类似 KeyBy 等分组聚合函数导致的，需要用户将热点 Key 进行预处理，降低或者消除热点 Key 的影。
GC：不合理的设置 TaskManager 的垃圾回收参数会导致严重的 GC 问题，我们
可以通过 -XX:+PrintGCDetails 参数查看 GC 的日志。
代码本身：开发者错误地使用 Flink 算子，没有深入了解算子的实现机制导致性能问题。我们可以通过查看运行机器节点的 CPU 和内存情况定位问题。
Flink中的状态存储
Flink在做计算的过程中经常需要存储中间状态，来避免数据丢失和状态恢复。选择的状态存储策略不同，会影响状态持久化如何和 checkpoint 交互。Flink 提供了三种状态存储方式：MemoryStateBackend、FsStateBackend、RocksDBStateBackend。
Operator Chains（算子链）这个概念你了解吗
为了更高效地分布式执行，Flink 会尽可能地将 operator 的 subtask 链接
（chain）在一起形成 task。每个 task 在一个线程中执行。将 operators 链接成 task 是非常有效的优化：它能减少线程之间的切换，减少消息的序列化/ 反序列化，减少数据在缓冲区的交换，减少了延迟的同时提高整体的吞吐量。这就是我们所说的算子链。
Flink的内存管理是如何做的
Flink 并不是将大量对象存在堆上，而是将对象都序列化到一个预分配的内存块上。此外，Flink大量的使用了堆外内存。如果需要处理的数据超出了内存限制，则会将部分数据存储到硬盘上。Flink 为了直接操作二进制数据实现了自己的序列化框架。
如何处理生产环境中的数据倾斜问题
flink数据倾斜的表现：
任务节点频繁出现反压，增加并行度也不能解决问题；
部分节点出现OOM异常，是因为大量的数据集中在某个节点上，导致该节点内存被爆，任务失败重启。
数据倾斜产生的原因：
业务上有严重的数据热点，比如滴滴打车的订单数据中北京、上海等几个城市的订单量远远超过其他地区；
技术上大量使用了 KeyBy、GroupBy 等操作，错误的使用了分组 Key，人为产生数据热点。
解决问题的思路：
业务上要尽量避免热点 key 的设计，例如我们可以把北京、上海等热点城市分成不同的区域，并进行单独处理；
技术上出现热点时，要调整方案打散原来的 key，避免直接聚合；此外 Flink 还提供了大量的功能可以避免数据倾斜。
Flink中的Time有哪几种
Flink中的时间有三种类型，如下图所示：
EventTime：是事件创建的时间。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink通过时间戳分配器访问事件时间戳。
Ingestion Time：是数据进入Flink的时间。
Processing Time：是每一个执行基于时间操作的算子的本地系统时间，与机器相关，默认的时间属性就是Processing Time。
例如，一条日志进入Flink的时间为2021-01-2210:00:00.123，到达Window的系统时间为2021-01-22 10:00:01.234，日志的内容如下：
2021-01-06 18:37:15.624 INFO Fail over to rm2
对于业务来说，要统计1min内的故障日志个数，哪个时间是最有意义的？—— eventTime，因为我们要根据日志的生成时间进行统计。
Flink对于迟到数据是怎么处理的
Flink中 WaterMark 和 Window 机制解决了流式数据的乱序问题，对于因为延迟而顺序有误的数据，可以根据eventTime进行业务处理，对于延迟的数据Flink 也有自己的解决办法，主要的办法是给定一个允许延迟的时间，在该时间范围内仍可以接受处理延迟数据
设置允许延迟的时间是通过allowedLateness(lateness: Time)设置
保存延迟数据则是通过sideOutputLateData(outputTag: OutputTag[T])保存
获取延迟数据是通过DataStream.getSideOutput(tag: OutputTag[X])获取文章推荐：
Flink 中极其重要的 Time 与 Window 详细解析
Flink中window出现数据倾斜怎么解决
window 产生数据倾斜指的是数据在不同的窗口内堆积的数据量相差过多。本质上产生这种情况的原因是数据源头发送的数据量速度不同导致的。出现这种情况一般通过两种方式来解决：
在数据进入窗口前做预聚合
重新设计窗口聚合的 key
Flink CEP编程中当状态没有到达的时候会将数据保存在哪里
在流式处理中，CEP 当然是要支持 EventTime 的，那么相对应的也要支持数据的迟到现象，也就是watermark的处理逻辑。CEP对未匹配成功的事件序列的处理，和迟到数据是类似的。在 Flink CEP的处理逻辑中，状态没有满足的和迟到的数据，都会存储在一个Map数据结构中，也就是说，如果我们限定判断事件序列的时长为5分钟，那么内存中就会存储5分钟的数据，这在我看来，也是对内存的极大损伤之一。
推荐阅读：一文学会Flink CEP
Flink设置并行度的方式
们在实际生产环境中可以从四个不同层面设置并行度：
操作算子层面(Operator Level)
.map(new RollingAdditionMapper()).setParallelism(10) //将操作算子设置并行度
执行环境层面(Execution Environment Level)
$FLINK_HOME/bin/flink 的-p参数修改并行度
客户端层面(Client Level)
env.setParallelism(10)
系统层面(System Level)
全局配置在flink-conf.yaml文件中，parallelism.default，默认是1：可以设置默认值大一点需要注意的优先级：算子层面>环境层面>客户端层面>系统层面。
Flink中Task如何做到数据交换
在一个 Flink Job 中，数据需要在不同的 task 中进行交换，整个数据交换是有 TaskManager 负责的，TaskManager 的网络组件首先从缓冲 buffer 中收集 records，然后再发送。Records 并不是一个一个被发送的，是积累一个批次再发送，batch 技术可以更加高效的利用网络资源。
Flink的内存管理是如何做的
Flink 并不是将大量对象存在堆上，而是将对象都序列化到一个预分配的内存块上。此外，Flink大量的使用了堆外内存。如果需要处理的数据超出了内存限制，则会将部分数据存储到硬盘上。Flink 为了直接操作二进制数据实现了自己的序列化框架。
介绍下Flink的序列化
Flink 摒弃了 Java 原生的序列化方法，以独特的方式处理数据类型和序列化，包含自己的类型描述符，泛型类型提取和类型序列化框架。
TypeInformation 是所有类型描述符的基类。它揭示了该类型的一些基本属性，并且可以生成序列化器。
TypeInformation 支持以下几种类型：
BasicTypeInfo: 任意 Java 基本类型或 String 类型
BasicArrayTypeInfo: 任意 Java 基本类型数组或 String 数组
WritableTypeInfo: 任意 Hadoop Writable 接口的实现类
TupleTypeInfo: 任意的 Flink Tuple 类型(支持 Tuple1 to Tuple25)。
Flink tuples 是固定长度固定类型的 Java Tuple 实现
CaseClassTypeInfo: 任意的 Scala CaseClass(包括 Scala tuples)
PojoTypeInfo: 任意的 POJO (Java or Scala)，例如，Java 对象的所有成员变量，要么是 public 修饰符定义，要么有 getter/setter 方法
GenericTypeInfo: 任意无法匹配之前几种类型的类
Flink海量数据高效去重
基于状态后端。
基于HyperLogLog：不是精准的去重。
基于布隆过滤器（BloomFilter）；快速判断一个key是否存在于某容器，不存在就直接返回。
基于BitMap；用一个bit位来标记某个元素对应的Value，而Key即是该元素。由于采用了Bit为单位来存储数据，因此可以大大节省存储空间。
基于外部数据库；选择使用Redis或者HBase存储数据，我们只需要设计好存储的Key即可，不需要关心Flink任务重启造成的状态丢失问题。
Flink SQL的是如何实现的
构建抽象语法树的事情交给了 Calcite 去做。SQLquery 会经过 Calcite 解析器转变成 SQL 节点树，通过验证后构建成 Calcite 的抽象语法树（也就是图中的 Logical Plan）。另一边，Table API 上的调用会构建成 Table API 的抽象语法树，并通过 Calcite 提供的 RelBuilder 转变成 Calcite 的抽象语法树。
然后依次被转换成逻辑执行计划和物理执行计划。
在提交任务后会分发到各个 TaskManager 中运行，在运行时会使用 Janino 编译器编译代码后运行。
第一时间获取最新大数据技术，尽在公众号：五分钟学大数据搜索公众号：五分钟学大数据，学更多大数据技术！其他大数据技术文档可下方扫码关注获取：
业务方面
在处理大数据过程中，如何保证得到期望值
保证在数据采集的时候不丢失数据，这个尤为重要，如果在数据采集的时候就已经不准确，后面很难达到期望值
在数据处理的时候不丢失数据，例如sparkstreaming处理kafka数据的时候，要保证数据不丢失，这个尤为重要
前两步中，如果无法保证数据的完整性，那么就要通过离线计算进行数据的校对，这样才能保证我们能够得到期望值
你感觉数仓建设中最重要的是什么
数仓建设中，最重要的是数据准确性，数据的真正价值在于数据驱动决策，通过数据指导运营，在一个不准确的数据驱动下，得到的一定是错误的数据分析，影响的是公司的业务发展决策，最终导致公司的策略调控失败。
数据仓库建模怎么做的
数仓建设中最常用模型–Kimball维度建模详解
数据质量怎么监控
单表数据量监控
一张表的记录数在一个已知的范围内，或者上下浮动不会超过某个阈值
SQL结果：var 数据量 = select count（*）from 表 where 时间等过滤条件
报警触发条件设置：如果数据量不在[数值下限, 数值上限]， 则触发报警
同比增加：如果((本周的数据量 -上周的数据量)/上周的数据量*100)不在 [比例下线，比例上限]，则触发报警
环比增加：如果((今天的数据量 - 昨天的数据量)/昨天的数据量*100)不在 [比例下线，比例上限]，则触发报警
报警触发条件设置一定要有。如果没有配置的阈值，不能做监控
日活、周活、月活、留存（日周月）、转化率（日、周、月）GMV（日、周、月）
复购率（日周月）
单表空值检测
某个字段为空的记录数在一个范围内，或者占总量的百分比在某个阈值范围内
目标字段：选择要监控的字段，不能选“无”
SQL结果：var 异常数据量 = select count(*) from 表 where 目标字段 is null
单次检测：如果(异常数据量)不在[数值下限, 数值上限]，则触发报警
单表重复值检测
一个或多个字段是否满足某些规则
目标字段：第一步先正常统计条数；select count(*) form 表；
第二步，去重统计；select count(*) from 表 group by 某个字段
第一步的值和第二步不的值做减法，看是否在上下线阀值之内
单次检测：如果(异常数据量)不在[数值下限, 数值上限]， 则触发报警
跨表数据量对比
主要针对同步流程，监控两张表的数据量是否一致
SQL结果：count(本表) - count(关联表)
阈值配置与“空值检测”相同
数据分析方法论了解过哪些？
数据商业分析的目标是利用大数据为所有职场人员做出迅捷，高质，高效的决策提供可规模化的解决方案。商业分析是创造价值的数据科学。
数据商业分析中会存在很多判断：
观察数据当前发生了什么？
比如想知道线上渠道A、B各自带来了多少流量，新上线的产品有多少用户喜欢，新注册流中注册的人数有多少。这些都需要通过数据来展示结果。
理解为什么发生？
我们需要知道渠道A为什么比渠道B好，这些是要通过数据去发现的。也许某个关键字带来的流量转化率比其他都要低，这时可以通过信息、知识、数据沉淀出发生的原因是什么。
预测未来会发生什么？
在对渠道A、B有了判断之后，根据以往的知识预测未来会发生什么。在投放渠道C、D的时候，猜测渠道C比渠道D好，当上线新的注册流、新的优化，可以知道哪一个节点比较容易出问题，这些都是通过数据进行预测的过程。
商业决策
所有工作中最有意义的还是商业决策，通过数据来判断应该做什么。这是商业分析最终的目的。 |
|
发表于 2022-9-21 21:08:31