如何用批处理实现禁用本地用户 Flink面试题

1. 为什么使用 Flink 替代 Spark？

使用 Flink 替代 Spark 的理由主要有以下几点：

低延迟、更吞吐量

Flink 是基于内存计算引擎设计的，因此具有比 Spark 更高的计算性能、吞吐量和低延迟。而且，在某些场景下，Flink 可以通过先进的任务调度和资源管理机制来优化作业的执行效率。

支持流处理和批处理

与 Spark 相比，Flink 的一个显著优势是同时支持流处理和批处理。这意味着对于只需要流处理或者仅需要批处理的应用，可以使用同一套系统来实现两种不同的需求。并且，Flink 的流处理模式采用了事件驱动的方式，可以更好地支持实时处理应用，很好地处理乱序数据。

更灵活、更易扩展

Flink 提供了比 Spark 更多的 API 和函数库，可以让开发人员更加自由地进行编程。同时，Flink 的作业调度器也更加灵活，可以在更大规模的数据中心部署和扩展，并且可以与其他分布式处理技术集成。

更强大的状态管理

Flink 的状态管理机制非常强大，可以轻松地处理存储在作业运行过程中产生的状态信息，从而保证作业在重启后能够正确恢复，保证 exactly-once 的状态一致

性，避免数据丢失和处理错误。

2. Flink 中filter、side output、split的区别？

filter：对原始流进行多次过滤，会降低性能

side output：Flink提供的最新的也是官方推荐的分流方法，可以进行多次拆分

split：使用split切分的流，不能进行二次切分，否则会报错

3. 说一下 Flink 状态后台管理机制？

Flink 提供了不同的状态后台管理机制来支持多种应用场景，包括本地内存、HDFS、RocksDB 等。

状态后台是指将流式应用中的状态保存到外部存储中，以避免因进程失败、任务重启等原因而导致的状态丢失。状态后台并不直接与应用程序交互，而是通过 Flink 运行时库提供的状态访问 API 与之交互。

Flink 提供了以下几种状态后台：

MemoryStateBackend：这是默认的状态后台，所有状态都保存在 TaskManager 的 JVM 堆内存中。如果作业失败或重启，所有状态都会丢失。

FsStateBackend：这个后台将状态保存在远程文件系统（如 HDFS）或本地文件系统中。在任务出现故障时，状态可以由恢复的任务重新加载并继续处理。

RocksDBStateBackend：当需要对非常大的键值状态进行处理时，这个后端通常表现最佳。它将状态存储在 RocksDB 中，该存储系统通过基于磁盘的哈希表实现快速键值存储和检索。同时，它也支持增量 checkpoint，可以在快速恢复应用程序的同时提供较高的写入吞吐量。

除了状态后台，Flink 还提供了 Checkpointing 机制来支持故障恢复。Checkpoint 会周期性地将应用程序状态保存到外部存储中如何用批处理实现禁用本地用户，并在失败时将其还原。

4. 怎么去重？考虑一个实时场景：双十一场景，滑动窗口长度为 1 小时， 滑动距离为 10 秒钟，亿级用户，怎样计算 UV？

在实时场景中，我们可以使用布隆过滤器来进行去重。布隆过滤器是一种数据结构，它可以快速判断一个元素是否存在于一个集合中。它通过多个独立的哈希函数将一个元素映射到一个固定长度的位图中，当需要判断某个元素是否在集合中时，会同时对这些位置上的比特位进行查询。如果所有的比特位都为 1，则认为该元素存在于集合中。

在双十一场景中，对于每个用户，我们可以将其用户 ID 使用多个不同的哈希函数进行哈希，得到多个哈希值，然后将这些哈希值对应的位设置为 1。当需要计算 UV 时，我们只需要统计滑动窗口内不同位的数量即可，这个数量就是近似的 UV 数量（可能存在少量误差）。由于哈希函数的性质，布隆过滤器有一定的误判率，但在亿级用户的场景下，误差通常可以接受。

具体实现上，可以使用 Flink 中提供的 BloomFilterStateDescriptor 将布隆过滤器序列化并存储在状态后端中，在每次处理输入数据时对布隆过滤器进行更新，并统计其中被设置为 1 的比特位数量即可。在实际场景中，为了提高精度和降低误判率，可以考虑使用多重哈希技术以及动态调整哈希函数数量和位图大小等优化策略。

5.Flink CEP 编程中当状态没有到达的时候会将数据保存在哪里？

在 Flink CEP 编程中，当某个事件符合某个规则的条件时，会将这个事件加入到一个状态中进行保存。如果在一段时间内该状态没有被满足，则 Flink CEP 将会自动清空该状态。

具体来说，Flink CEP 会将数据保存在分布式缓存或者磁盘中。在保存数据时，会基于规则和目前已经触发的事件，使用 Flink 内部的状态后端来维护相关的状态。这就意味着，在 Flink CEP 中处理的数据是有状态的如何用批处理实现禁用本地用户，并且需要确定一个恰当的生命周期来避免出现状态过期或缺失的情况。

同时，在 Flink CEP 中可以通过设置不同的清除策略来控制状态的清空。例如，可以根据时间、数量、大小等多种因素去自定义清空策略，以及对于特定的模式如何清空状态进行调整。

因此，当状态没有到达时，Flink CEP 会将数据保存在分布式缓存或者磁盘中，以便在后续的操作中进行参考和处理，以确保CEP 规则匹配状态的完整性和正确性。

6. 请说一下Flink 的 Watermark 机制？

Flink 中的 Watermark 是一种用于处理事件时间的延迟机制。在 Flink 中，每个数据流元素都会带有一个时间戳，这个时间戳可以是事件发生的时间（Event Time）或者数据进入 Flink 的时间（Ingestion Time）。为了正确处理 Event Time，Flink 引入了 Watermark 机制。

Watermark 是一种衡量 Event Time 进度的机制。它是一个特殊的时间戳，用来表示 Event Time 已经达到了某个时间点，已经没有更多的元素会再次返回该时间点之前的窗口。在 Flink 中，Watermark 可以通过自定义的 AssignerWithPeriodicWatermarks 或者 AssignerWithPunctuatedWatermarks 来生成。

AssignerWithPeriodicWatermarks 是定期地插入 Watermark，比如每秒插入一个 Watermark。而 AssignerWithPunctuatedWatermarks 是按需插入 Watermark，在某些条件下才会触发一个新的 Watermark。例如，当一个事件发生时，可以触发 AssignerWithPunctuatedWatermarks 插入一个 Watermark。

7.Flink 的 checkpoint 机制与spark的 checkpoint 机制比较？

Flink 和 Spark 都有 Checkpoint 机制，但是这两种框架的 Checkpoint 机制有所不同。

首先，Flink 的 Checkpoint 机制是一种轻量级的分布式快照技术每个算子都会保存自己的状态，并将其发送到专门的 Checkpoint 协调器（CheckpointCoordinator）节点上进行统一管理；而 Spark 的 Checkpoint 机制则是将 RDD 持久化到磁盘中，并且只在出现故障时恢复数据。因此，在大规模数据处理和高可靠性方面，Flink 的 Checkpoint 机制要比 Spark 更优秀。

其次，Flink 的 Checkpoint 机制比 Spark 更加灵活和可配置。例如，在 Flink 中，您可以通过更改 Checkpoint 间隔时间、最大并发 Checkpoint 数等参数来调整 Checkpoint 机制的执行方式，以适应不同的应用场景。而 Spark 的 Checkpoint 机制则没有太多的参数可以配置，因此灵活性略差。

另外，如果考虑到可伸缩性和吞吐量等方面，Flink 的 Checkpoint 机制也优于 Spark。这是因为 Flink 的 Checkpoint 机制可以支持任意大小的状态存储，而 Spark 的 Checkpoint 机制则需要将 RDD 存储到磁盘中，这可能会降低 Spark 应用程序的吞吐量和性能。

8. Flink 有哪些重启策略？

No Restart：不重启任务，一旦任务出现问题就直接 fail 掉，需要手动进行重启。

Fixed Delay Restart：固定时间间隔重启，当作业失败后，等待固定时间间隔之后自动重启，如果连续尝试重启次数超过阈值，则放弃重启。

Failure Rate Restart：根据失败率进行重启，在一段时间内观察上游数据源（source）或者下游数据接收器（sink）处理成功的比例，当超过预设的失败率后就会自动重启任务，否则放弃运行。

Full Restarts：无限制地重新启动任务，在遇到错误时一直重新启动任务，直到达到指定的最大尝试次数为止。这种重启策略主要用于测试和开发环境中。

9.Flink的内存管理

Flink 的内存管理可以分为三个部分：

TaskManager 内存

TaskManager 内存是 Flink 用于分配算子数据和状态的主要内存池。默认情况下，每个 TaskManager 的堆外内存最大占用量为 JVM 堆外空间的百分之十, 如果需要更多的内存，则需要使用堆外内存模式。此外，Flink 还提供了两种简单方式来控制内存使用：总内存大小限制和算子级别的内存限制。

Managed Memory

Managed Memory 是 Flink 中非常重要的一部分，主要用于存储运行时的状态信息，如排序、哈希表和窗口等算子状态。与 TaskManager 内存不同，Managed Memory 对外提供了更加细粒度的内存管理能力，其中包括了所有任务组件的内存分配、释放、回收操作。同时，在整个 FLINK 集群中也只有一个专门负责纯内存使用的管理线程（Memory Manager），它负责监控 Managed Memory 的使用情况，以及内存中的各个分区进行相互调度以达到最优化的状态。

Off-Heap Memory

除了上述两种类型的内存之外，Flink 允许使用 Off-Heap Memory （堆外内存）来处理特殊类型的数据。堆外内存的好处在于减少 GC 压力，从而提高了 Flink 处理大数据集的性能。堆外内存可以用来存储一些不易上下文关联和需要经常访问而不能使用 JVM 垃圾回收机制中的对象。

10. Flink中在使用聚合函数 GroupBy、Distinct、KeyBy 等函数时出现数据热点该如何解决？

在业务上规避问题：在一些场景中，我们可以通过对业务进行优化来减少或避免数据热点的出现。例如，在订单场景中，可以将北京和上海的订单拆分为多个子区域，然后对每个子区域进行聚合操作，最后再将结果合并。

Key 的设计上的优化：在实际应用中，某个键可能会含有过多与其他键不同的数据，造成数据倾斜，所以设计或合理选择键是很重要的。如在北京和上海的订单场景中，可以按地区、按时间段、按商品类别等拆分键进行聚合。

参数设置上的调整：在 Flink 1.9.0 SQL(Blink Planner) 中支持了 MiniBatch 技术，该技术会缓存一定量的数据后再触发处理，以减少对状态的访问。通过适当设置参数，可以在一定程度上减少数据热点问题带来的影响。

广播变量：使用广播变量将一些计算开销较小而数据量较小的信息复制到所有任务中。例如，在聚合时，可以将某些小规模的缓存数据广播给所有任务，而不需要在每个任务中对其进行计算。

需要根据具体场景和问题的实际情况选择合适的解决方案，同时可以结合以上几种方法进行使用，并尝试不断优化和调整以达到更好的性能效果。

11. Flink是如何处理反压的？

当数据来源的速度快于处理速度时，反压机制可以防止数据在系统中积压。Flink 采用的反压机制是异步的，基于下游操作符发送“背压信号”，如果上游操作符无法及时处理所有数据，则必须根据信号调整发送速率。

Flink 的反压机制可以通过以下方式来配置：

配置“最大并发任务数”以限制同时处理的任务数量。

调整检查点间隔（checkpoint interval）以增加或减少背压检测周期。

调整 TASK_MANAGER_NETWORK_REQUEST_BACKOFF_MAX 以增大后退时间，从而放慢数据源数据发送的速度。

在启用反压机制的情况下，Flink 可以通过向下游操作符发送背压信号来限制流速。如果下游操作符处理缓慢，Flink 将停止接收更多数据直到下游操作符能够处理更多数据为止。这种机制确保了系统的整体稳定性和可靠性，并防止数据积压和丢失。

12. Operator Chains（算子链）你了解吗？

Flink 的 Operator Chains（算子链）是一种性能优化技术，它可以将多个算子合并成一个算子链来执行，从而避免反复序列化和反序列化数据，减少网络传输和 I/O 操作的开销，提高数据处理的效率。

算子链中的所有算子都在同一个线程内执行，它们之间通过内部迭代器或本地缓存等方式进行数据交换，避免了线程切换和通信的开销。此外，在算子链中，由于每个算子都只处理一部分数据，因此可以利用容量更小的状态后端来保存算子状态，缓解状态管理的负担，提高算子的吞吐量和响应速度。

要创建算子链，需要调用StreamExecutionEnvironment中的enableOperatorChaining方法，并确保在构建算子时满足以下条件：

需要注意的是，虽然算子链可以有效地提高数据处理的效率，但它也存在一些限制和风险。如果算子链过长或者存在硬件资源瓶颈，会导致整个任务的延迟和吞吐量下降。同时，算子链中的任何一个算子出现错误都会导致整个算子链失败，因此需要做好异常处理和故障恢复等措施，保证数据处理的稳定性和正确性。

public class OperatorChainingExample {      public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.enableOperatorChaining(); // 启用算子链        DataStream input = env.socketTextStream("localhost", 9999); // 设置数据源        DataStream output = input.map(str -> str.toUpperCase()); // 使用 map 算子处理数据        output.print(); // 输出结果        env.execute("Operator Chaining Example"); // 启动任务    }}

在上面的示例中，我们通过调用enableOperatorChaining方法来启用算子链。然后，我们通过socketTextStream方法创建一个数据源，将数据流传递给map算子进行处理，最后调用print方法输出结果。这些算子将被自动组合成一个算子链并在同一个线程内执行。

13. Flink什么情况下才会把Operator chain在一起形成算子链？

14. flink消费kafka数据的时候，如何处理脏数据？

在 Flink 中处理 Kafka 数据时，脏数据（Dirty Data）通常是指无法被正确反序列化的消息。比如说，某些消息并不符合预定义的数据格式，或者某些消息字段存在错误等情况。针对这种情况，可以考虑以下两种方法来处理脏数据：

使用 Kafka Deserialization Schema 来过滤掉无法解析的消息。Deserialization Schema 用于将 Kafka 消息进行反序列化，如果某条消息无法被正确解析，则会被丢弃。因此，在实现自定义的 Deserialization Schema 时，需要注意对异常输入流进行处理，避免程序崩溃。

在 Flink 应用程序中使用侧输出流（Side Output）来收集脏数据。当某条消息无法被正确解析时，可以通过侧输出流输出该条消息，便于后续进行人工处理或其他特殊逻辑处理。

15. Flink的Savepoint和Checkpoint是什么关系？

在 Flink 中，Savepoint 和 Checkpoint 是两个相关但是不同的概念。

Checkpoint 是指定时间系统对分布式流应用程序状态进行一致性检查的机制。Checkpoint 机制旨在为故障恢复提供支持，通过将整个系统的状态存储在可靠的持久化存储中，以便在发生故障时可以重新启动并从之前的状态恢复。

Savepoint 则是 Flink 中一种特殊的 checkpoint（检查点），它允许在保存点创建时将应用程序状态快照保存到持久化存储中。通过使用 Savepoint，可以实现在 Flink 应用程序升级、回滚、故障恢复等场景中存储应用程序状态。

可以看出，Checkpoint 和 Savepoint 的目的都是为了应对可能出现的故障情况，但两者的作用方式和应用场景有所不同。Checkpoint 是周期性地记录 Flink 应用程序的状态，而 Savepoint 是在特定时间点手动触发，并将应用程序的状态快照保存到持久化存储中。

此外，在 Flink 中，当使用 Savepoint 时，会自动创建一个 checkpoint。这意味着，在发生故障时，可以同时使用这两个机制来实现应用程序的恢复。具体而言，可以将 Flink 容错语义设置成 EXACTLY_ONCE，然后在每个 checkpoint 完成时同时创建一个 Savepoint，以便在发生故障时可以从最近的 Savepoint 开始恢复。