doris的分区规则

为什么表结构的分区与分区字段里去重值数量不一样？

How to Avoid the ReceiverDisconnectedException

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

21/08/19 02:18:45 ERROR TaskSetManager [task-result-getter-0]: Task 32 in stage 73.0 failed 4 times; aborting job
21/08/19 02:18:45 ERROR FileFormatWriter [stream execution thread for [id = 5f382b82-2006-49a4-a657-5512d5bc6cce, runId = e783c09d-8ea7-4848-b018-2422590426ce]]: Aborting job 6d36e286-ecb0-4a8e-9928-5ac7a0fbb7d2.
org.apache.spark.SparkException: Job aborted due to stage failure: Task 32 in stage 73.0 failed 4 times, most recent failure: Lost task 32.3 in stage 73.0 (TID 18190, wn126-msnbi.awfbdxsze1iudhhki0l2sbzfaf.bx.internal.cloudapp.net, executor 8): java.util.concurrent.CompletionException: com.microsoft.azure.eventhubs.ReceiverDisconnectedException: New receiver 'spark-26-18237' with higher epoch of '0' is created hence current receiver 'spark-8-18190' with epoch '0' is getting disconnected. If you are recreating the receiver, make sure a higher epoch is used. TrackingId:ddef38220000007a00008bf9611dbf7b_G4_B16, SystemTracker:msnsam-prod:eventhub:sambeacon-ueq~4223|$default, Timestamp:2021-08-19T02:18:45, errorContext[NS: msnsam-prod.servicebus.windows.net, PATH: sambeacon-ueq/ConsumerGroups/$Default/Partitions/32, REFERENCE_ID: LN_23ed71_1629339510773_d5c9_G4, PREFETCH_COUNT: 500, LINK_CREDIT: 200, PREFETCH_Q_LEN: 0]
	at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292)
	at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:308)
	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:607)
	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
	at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1990)
	at com.microsoft.azure.eventhubs.impl.ExceptionUtil.completeExceptionally(ExceptionUtil.java:116)
	at com.microsoft.azure.eventhubs.impl.MessageReceiver.drainPendingReceives(MessageReceiver.java:505)
	at com.microsoft.azure.eventhubs.impl.MessageReceiver.onError(MessageReceiver.java:490)
	at com.microsoft.azure.eventhubs.impl.MessageReceiver.onClose(MessageReceiver.java:790)
	at com.microsoft.azure.eventhubs.impl.BaseLinkHandler.processOnClose(BaseLinkHandler.java:73)
	at com.microsoft.azure.eventhubs.impl.BaseLinkHandler.handleRemoteLinkClosed(BaseLinkHandler.java:109)
	at com.microsoft.azure.eventhubs.impl.BaseLinkHandler.onLinkRemoteClose(BaseLinkHandler.java:51)
	at org.apache.qpid.proton.engine.BaseHandler.handle(BaseHandler.java:176)
	at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:108)
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.dispatch(ReactorImpl.java:324)
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:291)
	at com.microsoft.azure.eventhubs.impl.MessagingFactory$RunReactor.run(MessagingFactory.java:784)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: com.microsoft.azure.eventhubs.ReceiverDisconnectedException: New receiver 'spark-26-18237' with higher epoch of '0' is created hence current receiver 'spark-8-18190' with epoch '0' is getting disconnected. If you are recreating the receiver, make sure a higher epoch is used. TrackingId:ddef38220000007a00008bf9611dbf7b_G4_B16, SystemTracker:msnsam-prod:eventhub:sambeacon-ueq~4223|$default, Timestamp:2021-08-19T02:18:45, errorContext[NS: msnsam-prod.servicebus.windows.net, PATH: sambeacon-ueq/ConsumerGroups/$Default/Partitions/32, REFERENCE_ID: LN_23ed71_1629339510773_d5c9_G4, PREFETCH_COUNT: 500, LINK_CREDIT: 200, PREFETCH_Q_LEN: 0]
	at com.microsoft.azure.eventhubs.impl.ExceptionUtil.toException(ExceptionUtil.java:43)
	at com.microsoft.azure.eventhubs.impl.MessageReceiver.onClose(MessageReceiver.java:789)
	... 15 more

Driver stacktrace:
	at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1889)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1877)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1876)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
	at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1876)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:926)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:926)
	at scala.Option.foreach(Option.scala:257)
	at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:926)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2110)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2059)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2048)
	at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
	at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:737)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2065)
	at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:167)
	at org.apache.spark.sql.execution.streaming.FileStreamSink.addBatch(FileStreamSink.scala:131)
	at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$runBatch$5$$anonfun$apply$17.apply(MicroBatchExecution.scala:537)
	at org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:78)
	at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125)
	at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73)
	at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$runBatch$5.apply(MicroBatchExecution.scala:535)
	at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:351)
	at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:58)
	at org.apache.spark.sql.execution.streaming.MicroBatchExecution.org$apache$spark$sql$execution$streaming$MicroBatchExecution$$runBatch(MicroBatchExecution.scala:534)
	at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply$mcV$sp(MicroBatchExecution.scala:198)
	at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:166)
	at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:166)
	at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:351)
	at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:58)
	at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1.apply$mcZ$sp(MicroBatchExecution.scala:166)
	at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:56)
	at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:160)
	at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:281)
	at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:193)

flink1.6

官网链接

Keyed Windows

1
2
3
4
5
6
7
8
9

stream
       .keyBy(...)               <-  keyed versus non-keyed windows
       .window(...)              <-  required: "assigner"
      [.trigger(...)]            <-  optional: "trigger" (else default trigger)
      [.evictor(...)]            <-  optional: "evictor" (else no evictor)
      [.allowedLateness(...)]    <-  optional: "lateness" (else zero)
      [.sideOutputLateData(...)] <-  optional: "output tag" (else no side output for late data)
       .reduce/aggregate/fold/apply()      <-  required: "function"
      [.getSideOutput(...)]      <-  optional: "output tag"

Non-Keyed Windows

1
2
3
4
5
6
7
8

stream
       .windowAll(...)           <-  required: "assigner"
      [.trigger(...)]            <-  optional: "trigger" (else default trigger)
      [.evictor(...)]            <-  optional: "evictor" (else no evictor)
      [.allowedLateness(...)]    <-  optional: "lateness" (else zero)
      [.sideOutputLateData(...)] <-  optional: "output tag" (else no side output for late data)
       .reduce/aggregate/fold/apply()      <-  required: "function"
      [.getSideOutput(...)]      <-  optional: "output tag"

窗口生命周期

window

trigger

The function will contain the computation to be applied to the contents of the window, while the Trigger specifies the conditions under which the window is considered ready for the function to be applied. A triggering policy might be something like “when the number of elements in the window is more than 4”, or “when the watermark passes the end of the window”. A trigger can also decide to purge a window’s contents any time between its creation and removal. Purging in this case only refers to the elements in the window, and not the window metadata. This means that new data can still be added to that window.

function包含了窗口中数据的计算逻辑，trigger指定了在哪些条件下会调用这些function

Evictor

Apart from the above, you can specify an Evictor (see Evictors) which will be able to remove elements from the window after the trigger fires and before and/or after the function is applied.

Flink的DataSource三部曲之一：直接API

Flink的DataSource三部曲之二:内置connector

Flink的DataSource三部曲之三:自定义

• WindowAssigner
  • SlidingProcessingTimeWindows
  • BaseAlignedWindowAssigner
    • SlidingAlignedProcessingTimeWindows
  • TumblingEventTimeWindows
    • TumblingTimeWindows
  • MergingWindowAssigner
    • ProcessingTimeSessionWindows
    • DynamicProcessingTimeSessionWindows
    • DynamicEventTimeSessionWindows
    • EventTimeSessionWindows
  • TumblingProcessingTimeWindows
  • SlidingEventTimeWindows
    • SlidingTimeWindows
  • GlobalWindows

WindowAssigner

1. SlidingProcessingTimeWindows

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

/**
 * A {@link WindowAssigner} that windows elements into sliding windows based on the current
 * system time of the machine the operation is running on. Windows can possibly overlap.
 *
 * <p>For example, in order to window into windows of 1 minute, every 10 seconds:
 * <pre> {@code
 * DataStream<Tuple2<String, Integer>> in = ...;
 * KeyedStream<String, Tuple2<String, Integer>> keyed = in.keyBy(...);
 * WindowedStream<Tuple2<String, Integer>, String, TimeWindows> windowed =
 *   keyed.window(SlidingProcessingTimeWindows.of(Time.of(1, MINUTES), Time.of(10, SECONDS));
 * } </pre>
 */

/**
 * 基于flink处理时间的滑动窗口
 */

6. SlidingEventTimeWindows

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

/**
 * A {@link WindowAssigner} that windows elements into sliding windows based on the timestamp of the
 * elements. Windows can possibly overlap.
 *
 * <p>For example, in order to window into windows of 1 minute, every 10 seconds:
 * <pre> {@code
 * DataStream<Tuple2<String, Integer>> in = ...;
 * KeyedStream<Tuple2<String, Integer>, String> keyed = in.keyBy(...);
 * WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed =
 *   keyed.window(SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(10)));
 * } </pre>
 */

/**
 * 基于元素时间的滑动窗口
 */

6.1 SlidingTimeWindows(废弃)

2. BaseAlignedWindowAssigner

1.1 SlidingAlignedProcessingTimeWindows

5. TumblingProcessingTimeWindows

3. TumblingEventTimeWindows

1
2
3
4
5
6
7
8
9
10
11
12

/**
 * A {@link WindowAssigner} that windows elements into windows based on the timestamp of the
 * elements. Windows cannot overlap.
 *
 * <p>For example, in order to window into windows of 1 minute:
 * <pre> {@code
 * DataStream<Tuple2<String, Integer>> in = ...;
 * KeyedStream<Tuple2<String, Integer>, String> keyed = in.keyBy(...);
 * WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed =
 *   keyed.window(TumblingEventTimeWindows.of(Time.minutes(1)));
 * } </pre>
 */

3.1 TumblingTimeWindows(废弃)

1
2
3
4
5
6

/**
 * A {@link WindowAssigner} that windows elements into windows based on the timestamp of the
 * elements. Windows cannot overlap.
 *
 * @deprecated Please use {@link TumblingEventTimeWindows}.
 */

4. MergingWindowAssigner

4.1 ProcessingTimeSessionWindows

4.2 DynamicProcessingTimeSessionWindows

4.3 DynamicEventTimeSessionWindows

4.4 EventTimeSessionWindows

7. GlobalWindows

总结

适用场景

1. 滑动窗口

• 每条数据会发送到多个滑动窗口中，即在最终的输出中，一条数据要被统计多次
• 适合统计据当前时间往前一段时间内的数据汇总

2. 滚动窗口

• 每条数据只会在一个滚动窗口中
• 适合对数据进行简单聚合后，再次聚合的场景
• 适合输出明细，不做聚合的场景，比如join后直接输出

占用内存比较

1. 滑动窗口
   • 理论上一条数据会复制到多个窗口，被复制几次，占用内存就会扩大几倍，但不清楚是否有优化，比如只复制数据的引用？
2. 滚动窗口

关于时间点

窗口的开始时间

1. 首先要明确的是，开始时间只与system time和offset参数相关，与程序开始运行时间无关
2. 比如设置了窗口size是1h，那么在

窗口的结束时间

因为本地测试中，本地构建的数据源并非预想中的单线程运行，所以想看下线程id，以确定是否每次都新起了线程。

推荐使用xml，结构清晰

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
    <Appenders>
        <!-- 控制台输出 -->
        <Console name="Console" target="SYSTEM_OUT">
            <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
        </Console>

        <!-- 文件输出（可选） -->
        <File name="File" fileName="logs/flink-job.log">
            <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
        </File>
    </Appenders>

    <Loggers>
        <!-- Flink 相关日志级别 -->
        <Logger name="org.apache.flink" level="INFO"/>

        <!-- 你的项目包名 -->
        <Logger name="com.jingqicao.code" level="DEBUG"/>

        <Root level="INFO">
            <AppenderRef ref="Console"/>
            <AppenderRef ref="File"/>
        </Root>
    </Loggers>
</Configuration>

解析

如果Logger不指定additivity=false，默认为true。此时当前Logger及Root都会处理日志，如果Logger与Root配了相同的Appender，就会重复输出。

Root的level，表示的是Root使用的日志级别，如果有一个Logger没有具体指定，就会交由Root处理，即日志级别由Root决定。

⚙️ 原理简述

• Flink 为许多常见 Scala 类型（如基本类型、集合、元组、case class 等）提供了内置的 TypeInformation 工厂。
• createTypeInformation 结合 ClassTag 和这些工厂，在编译期生成正确的 TypeInformation，避免了手动构造的繁琐和错误。

⚠️ 注意事项

• 必须导入：如果忘记导入 org.apache.flink.streaming.api.scala._，编译器会报错“无法找到隐式值”，提示需要 TypeInformation。
• 自定义类型：对于自定义的 case class 或类，只要其成员类型都有可用的 TypeInformation，createTypeInformation 也能自动组合生成。
• Table API 中也有类似机制，但通常通过导入 org.apache.flink.table.api.bridge.scala._ 来提供。

[toc]

保存算法

checkpoint保存算法有几种

Flink的Checkpoint机制是其容错和状态一致性的核心。根据触发机制、数据保存方式和处理逻辑的不同，Checkpoint的实现方式可以从几个不同的维度进行分类。

🔀 按屏障对齐方式分类

从Flink 1.13开始，官方引入了超时对齐机制，即“对齐超时后自动切换为非对齐”的混合模式，以平衡两种方式的优缺点-。

📊 按状态备份方式分类

⚡ 按执行阶段分类

🚀 高级优化：通用增量检查点 (Generic Log-Based Incremental Checkpoint)

在Flink 1.15引入，1.16达到生产可用-1-18

这是一种更先进的增量Checkpoint架构，主要优化包括：

• 核心机制：引入State Changelog机制，将状态更新的变化日志持续写入远程存储（如HDFS/S3），而Checkpoint仅需Flush这些日志-1-18。
• 异步优化：将状态持久化与Checkpoint过程完全解耦，避免大文件上传的不确定性-18。
• 主要优势：Checkpoint完成时间更稳定、快速，端到端延迟更低-1-18。

💎 总结

Flink的Checkpoint算法可以根据不同维度进行分类：

• 按屏障对齐方式：分为对齐和非对齐两种，主要权衡反压场景下的Checkpoint完成速度与额外的存储开销。
• 按状态备份方式：分为全量和增量两种，增量方式极大地优化了大状态作业的Checkpoint性能。
• 按执行阶段：分为同步和异步两个阶段，将耗时的持久化操作异步化，减少对数据处理路径的影响。
• 高级优化：通用增量Checkpoint（1.15+）是面向未来的架构，通过解耦Checkpoint与状态存储，实现了更稳定高效的Checkpoint。

这些算法通常组合使用。例如，一个生产作业可能配置为：使用RocksDB状态后端、开启增量Checkpoint、并设置对齐超时自动切换为非对齐。

对齐算法中 Barrier对齐期间会阻塞后续数据处理 是什么意思

简单来说，“Barrier对齐期间会阻塞后续数据处理” 指的是Flink为了保证数据一致性，在Exactly-Once模式下执行检查点时，会采取的一种“停下来等”的策略。这个策略确实会导致数据处理出现短暂的“暂停”，以换取状态快照的绝对精准。

⏸️ 阻塞场景：为什么必须“等”？

要理解阻塞，需要先了解Flink的Barrier对齐机制。当JobManager触发一次Checkpoint时，Source算子会向数据流中注入带有ID的Barrier（屏障）-11。对于只有一个输入流的算子，Barrier的处理相对简单，它会在收到Barrier时直接触发快照-23。

阻塞主要发生在**有多个输入流的算子（如CoProcessFunction, Join等）**上-23。因为不同的输入流数据到达速度可能不同，Barrier到达的时间也会有先后-。

为了保证快照能精准地反映Barrier到达前那一刻所有输入流的状态，算子必须执行“对齐”操作：

1. 先到的先等：当算子从某个输入流（比如输入流A）收到了一个Barrier（编号为n）时，它立即停止处理该输入流Barrier之后的任何新数据-23。
2. 继续处理其他：算子会继续处理其他尚未收到Barrier的输入流（比如输入流B）中的数据-11。
3. 对齐完成：直到输入流B的Barrier（编号也是n）也到达后，所有输入流的Barrier才算“对齐”-11。这时，算子才会触发自己的状态快照，并向下游广播Barrier-23。

这个过程中，从第一个Barrier到达，到最后一个Barrier到达的这段时间，就是“对齐时间”（Alignment Duration）-2-。在此期间，输入流A的处理被阻塞，输入流A后续的数据会被暂存在一个缓冲区里，等待对齐完成后才被继续处理-23-25。如果这个过程非常耗时，就说明作业可能正经历反压-2。

🆚 两种语义：阻塞与不阻塞的区别

这种“阻塞”行为，是实现 Exactly-Once 的关键，它能确保所有输入流的状态在快照时是完全一致的。

如果为了追求低延迟而选择“不等待”，即一旦接收到任何一个Barrier就立即触发快照并继续处理所有数据，这就变成了 At-Least-Once 语义。这样虽然减少了阻塞，但可能在恢复时造成数据重复处理-7。

💡 阻塞的影响：反压与性能

“阻塞”虽然保证了准确性，但也可能带来一些副作用：

• 加剧反压：在数据倾斜或下游处理能力不足时，对齐过程中的等待会进一步减慢数据处理速度，导致Barrier流动缓慢，Checkpoint耗时变长-11。
• 增加内存开销：用于暂存阻塞数据的输入缓冲区会消耗内存。如果对齐时间过长，缓冲区积压的数据会增多，可能导致内存压力过大-23。

🔧 解决方案：非对齐检查点 (Unaligned Checkpoint)

为了解决“对齐阻塞”在反压场景下带来的性能问题，Flink从1.11版本开始引入了 非对齐检查点 (Unaligned Checkpoint) -7。

它的核心思路是 “不停下来等，而是把等待时正在传输的数据也保存起来”。当算子的某个输入流收到Barrier后，它不再阻塞该通道，而是立即开始做快照，并将所有输入和输出缓冲区中“正在飞行”的数据作为快照的一部分一起持久化-11。这相当于把对齐的压力从运行时转移到了恢复时，有效避免了因对齐阻塞而加剧反压的问题-11。

💎 总结

“Barrier对齐阻塞”是Flink为换取Exactly-Once语义而设计的机制，它通过在快照制作时暂停部分数据处理来确保状态的一致性。这个机制在反压时可能成为瓶颈，而非对齐检查点正是为了解决这一问题而生的。

书中exactly once语义一节中看到一句话，其中一个步骤是算子任务对状态做快照，书上说，当分界线传递到了内部的window()算子，它同样会对自己的状态进行快照保存，并写入远程的持久化存储中。状态后端不是有两种吗，其中一种是写入内存的，为什么这里直接说写入持久化存储了？

这个观察很敏锐。书中的描述其实是从实现原理的角度出发的。对于Flink来说，只要开启了Checkpoint，即使使用看起来只存在内存中的MemoryStateBackend，状态也会以文件形式被持久化到远程存储。这其实是Flink实现Exactly-Once语义的基石。

MemoryStateBackend这个名称确实容易引起误解，它指的主要是运行时状态在内存中，但Checkpoint数据依然需要可靠的持久化。

🧐 一个名字引发的误解：MemoryStateBackend

根据Flink官方文档，MemoryStateBackend的核心行为是：

• 运行时状态：保存在TaskManager的JVM堆内存中-1。
• Checkpoint时：状态数据被发送到JobManager，并为了高可用性（HA）被持久化到文件系统中-1。它本质上是一个可以脱离文件系统工作的“基于文件系统的后端”-1。

这种设计在Flink 1.12的官方文档里说得更直接：MemoryStateBackend在Checkpoint时，“将快照信息作为CheckPoint应答消息的一部分发送给JobManager(master)，同时JobManager也将快照信息存储在堆内存中”-5。这里并没有提及直接持久化到远程存储，所以书里的说法，很可能是基于更高版本中对其HA机制的描述，或是从“实现Exactly-Once语义”这个更高阶的视角出发，将Checkpoint视为最终必须写入持久化存储的行为。

💡 概念重塑：状态后端与检查点存储的“解耦”

Flink 1.15及之后版本的重要演进，可以帮助我们更好地理解这种设计：Flink将状态后端（State Backend）和检查点存储（Checkpoint Storage）这两个概念彻底分开了-3。这意味着，你完全可以组合使用不同的后端和存储方式：

• HashMapStateBackend (等价于MemoryStateBackend) 处理运行时状态。
• JobManagerCheckpointStorage 负责将Checkpoint数据存储在JobManager内存中，适用于本地测试-3。

但为了实现生产级的容错（Exactly-Once语义），通常会将HashMapStateBackend与FileSystemCheckpointStorage搭配使用，这样状态快照就会被可靠地写入HDFS或S3等分布式文件系统-3。

⚙️ 三种官方状态后端如何工作

下图清晰地展示了Flink官方提供的三种状态后端，在工作状态存储和Checkpoint持久化上的分工：

💎 总结

所以，书上的那句话并没有错。它强调的是Exactly-Once语义实现的关键一环——将全局一致的状态快照进行持久化存储，而具体存到哪里，则由所选的状态后端和检查点存储配置共同决定。MemoryStateBackend虽然运行时状态在内存中，但为了实现生产级的容错，它的Checkpoint数据最终也需要被写入可靠的存储中。

也就是说，当开启了checkpoint，状态可能会存两份，一份在内存，一份要持久化？

是的，你的理解非常准确。开启 Checkpoint 后，Flink 的状态确实会存在“两份”：一份是用于处理实时数据的工作状态（Working State），另一份是用于故障恢复的快照状态（Snapshot State）。

这“两份”数据虽然内容上有关联，但它们的存储位置、生命周期和作用是完全不同的。

两份状态的详细对比

一个类比：文档编辑的“自动保存”

你可以把它们想象成正在编辑一份文档：

• 工作状态 就是你屏幕上正在编辑的文档内容，你所有的修改都在这里实时发生。它就相当于保存在你电脑内存/缓存里的“工作状态”。
• 快照状态 就是你每隔一段时间点击“保存”后，存到硬盘上的那个文件。那个文件是过去某个时间点的静态副本，主要用于系统崩溃后恢复。

而开启 Checkpoint 后的过程就像这样：

1. 你的文档（工作状态）一直在内存中，你可以随时编辑（处理数据）。
2. 每隔几分钟，系统自动帮你把当前文档内容完整地存一次盘（生成快照）。
3. 你的电脑内存里有一份正在编辑的文档（工作状态），硬盘上则可能保留着最近几次存盘的副本（快照状态）。

两点补充说明

1. “两份”不等于“双倍”开销
   • 虽然数据是两份，但生成快照的过程（特别是异步和增量快照）已经高度优化，通常不会让数据处理的延迟翻倍。
   • 增量快照每次只保存自上次快照以来的变化部分，而不是全部数据，这大大减少了快照的数据量和网络传输。
2. 不是所有状态后端都“在内存”
   • 如果使用 RocksDBStateBackend，工作状态本身主要存储在本地磁盘（RocksDB），而不仅仅是内存。这时，“两份”的含义就变成了：一份在本地磁盘（工作状态），另一份在远程存储（快照）。

总结

所以，你的理解完全正确：Checkpoint 机制本质上就是用远程存储的静态快照，为本地运行的动态工作状态提供了一层可靠的容错保障。这是一套非常经典的“动态数据 + 静态备份”的分布式系统设计模式。

[toc]

Map[String, () => Solution]

以函数作为map值存储的

在 Scala 中，Map[String, () => Solution] 里的 () => Solution 是一个函数类型，它表示一个不接受任何参数、返回一个 Solution 对象的函数。将这样的函数作为 Map 的值存储，体现了 函数作为一等公民（first-class citizen） 的特性，即函数可以像普通值一样被传递、存储和调用。

具体来说，这里的 () => Solution 是一个函数值（function value），它是函数类型的实例。你可以把它看作是一个工厂：当你需要一个新的 Solution 实例时，就调用这个函数（通过 apply() 或直接加括号 ()），它会生成一个实例。这种存储函数而不是直接存储实例的做法，本质上是在存储一个生产实例的行为，而不是实例本身。

这种设计模式常被称为工厂函数存储或延迟初始化，它是函数式编程中常见的技巧，利用了函数作为值的灵活性。

函数的返回值

根据返回值数量选择合适的数据类型

固定数量、相同类型 → 元组 / case class

可变数量、相同类型 → 集合（List、Seq、Vector 等）

可变数量、不同类型 → 用 sealed trait 和集合模拟，或使用 HList（shapeless）

两种可能的结果 → Either、Try、Option

当返回值可能成功也可能失败时：使用 Either 或 Try

如果你的函数可能返回多种结果（例如成功时返回一个值，失败时返回错误信息），可以使用 Either。

1
2
3

def divide(a: Int, b: Int): Either[String, Int] = 
  if (b == 0) Left("Division by zero")
  else Right(a / b)

Either 通常用于表示两种可能的结果，但你可以嵌套组合来表示更多情况。

富函数

为什么可以通过富函数类来自定义按键分区状态

在 Flink 中，富函数（RichFunction） 能够自定义按键分区状态（Keyed State），是因为 Flink 的设计将 “运行时上下文（RuntimeContext）” 和 “状态后端（State Backend）” 无缝集成到了富函数的生命周期中。

简单来说：富函数充当了用户逻辑与 Flink 分布式状态管理机制之间的桥梁。下面从几个关键点解释其原理。

1. 富函数提供 RuntimeContext，这是状态管理的入口

当你在 KeyedStream 上使用富函数（如 RichFlatMapFunction、KeyedProcessFunction）时，Flink 会在算子初始化时通过 open() 方法将 RuntimeContext 注入到函数实例中。

RuntimeContext 提供了以下关键方法，用于创建和管理按键分区状态：

• getState(ValueStateDescriptor)
• getListState(ListStateDescriptor)
• getMapState(MapStateDescriptor)
• getReducingState(ReducingStateDescriptor)
• getAggregatingState(AggregatingStateDescriptor)

这些方法返回的状态对象（如 ValueState）会自动绑定到当前处理的数据的 key 上。

2. 按键分区状态的自动 key 感知

按键分区状态的核心特点是：每个不同的 key 都有自己独立的状态副本。富函数在 KeyedStream 上使用时，Flink 的运行时框架会自动将当前处理的元素的 key 与状态对象关联。

具体流程：

• 当数据进入算子时，Flink 知道这条数据属于哪个 key（通过 KeySelector 或分区方式确定）。
• 在调用 processElement() 等方法时，RuntimeContext 内部会根据当前 key 找到对应的状态分区。
• 用户代码通过状态对象读写时，实际上操作的是当前 key 对应的状态，无需手动管理 key 到状态的映射。

这种机制被称为 “按键分区状态的自动作用域（scoping）”。

3. 状态后端负责实际存储和容错

用户通过富函数定义的状态描述符（StateDescriptor）只声明了状态的名称、类型、序列化器以及可选的TTL（Time-To-Live）。真正的状态数据由 状态后端（State Backend） 管理：

• MemoryStateBackend：状态存储在 JVM 堆内存中，适合本地测试。
• FsStateBackend：状态存储在内存中，但检查点会持久化到文件系统。
• RocksDBStateBackend：状态存储在本地 RocksDB 数据库中，支持超大状态。

无论使用哪种状态后端，富函数中的状态操作都通过统一的 State 接口访问，Flink 会在后台处理序列化、缓存、磁盘读写、检查点快照等复杂逻辑。

4. 容错与一致性保证

富函数中的按键分区状态会自动参与检查点（Checkpoint） 机制。当作业失败恢复时，Flink 会从最近成功的检查点中恢复所有 key 的状态，并将其重新加载到对应的算子实例中。这一过程对用户代码完全透明，只需在富函数中定义状态即可。

5. 设计优势：为什么不用普通函数？

如果不使用富函数，而是自己维护一个 Map<Key, Value> 来实现类似功能，会遇到以下问题：

• 内存溢出：状态无法自动溢写到磁盘。
• 故障恢复困难：需要手动实现状态的快照和恢复。
• 无法感知 key 的变化：需要手动从数据中提取 key 并维护映射。
• 性能差：缺少 Flink 内部的优化（如状态缓、增量快照、异步检查点等）。

富函数 + 按键分区状态的设计，将分布式状态管理的复杂性完全封装，让用户能像写单机程序一样使用状态，同时享受 Flink 提供的高性能、一致性和容错能力。

💎 总结

因此，通过富函数自定义按键分区状态，本质上是 Flink 将分布式状态管理抽象为易用的编程模型，让开发者无需关心底层的 key 路由、状态分区、容错快照等细节。

富函数只能在KeyedStream上使用吗

不是的，富函数（RichFunction）并不仅限于 KeyedStream，它可以在任何数据流上使用。

几乎所有 Flink 算子（如 map, filter, flatMap 等）都同时提供了普通函数和对应的富函数版本，例如 RichMapFunction, RichFilterFunction, RichFlatMapFunction-20-8。即使是在非按键分区的普通数据流（DataStream）上，你依然可以无缝地使用这些富函数-。

🧩 富函数在非KeyedStream上的主要应用

虽然不能使用键控状态（Keyed State），但在非KeyedStream上，富函数依然提供了两个非常重要的能力：

• 生命周期管理 (Lifecycle Management)：RichFunction 提供了 open() 和 close() 两个生命周期方法-36。
  • open()：在算子的实际工作方法（如 map()）被调用之前执行一次，常用于一次性初始化任务，如建立数据库连接、加载配置文件等-20。
  • close()：在算子生命周期结束时调用，用于执行清理工作，如关闭连接、释放资源等-20。
• 获取运行时上下文 (Access to RuntimeContext)：通过 getRuntimeContext() 方法，可以获取算子的运行时信息，例如算子并行度、任务名称等-36。这为开发通用、自适应的函数提供了基础。

🚫 在非KeyedStream上的能力限制

在非 KeyedStream 上使用富函数时，确实无法访问以下两项关键功能：

• 无法访问键控状态 (Keyed State)：键控状态是基于 Key 进行分区的，由于 DataStream 没有经过 keyBy() 分区，因此无法获取和使用 ValueState, ListState 等键控状态--5。
• 无法访问定时器服务 (TimerService)：TimerService（如 registerEventTimeTimer()）依赖于为特定 Key 注册定时器的能力，因此在非 KeyedStream 上不可用--1。

💎 总结

• 能否使用？ 可以，富函数可以用在任何数据流上。
• 与 KeyedStream 的区别？ 非 KeyedStream 上无法使用键控状态和定时器服务-2，但可以正常使用生命周期方法和运行时上下文。