Queryable Timestamp

难以调和的三者：数据新鲜度、查询性能和资源消耗

Napa主架构

Data Availability

作者：肯尼思·内尔·丘基尔、维克托·梅耶—舍恩伯格

2013-07-29 09:53:56

每个人都知道互联网改变了企业经营、政府运作以及人们生活的方式。但是一种新的、不那么明显的技术趋势却有着同样巨大的变革能力，那就是“大数据”。大数据的趋势发端于下面这个事实：如今到处传播的信息比以往任何时候都多出了许多，而且这一趋势正在应用于非同寻常的新用途。大数据与互联网截然不同，虽然互联网使数据的收集和共享方便了很多。大数据的意义并不仅仅是通信：其本质是我们可以从大量的信息中学习到从较少量的信息中无法获取的东西。

0.1 将改变人类思考方式

早在公元前3世纪，亚历山大图书馆被认为收藏了全部的人类知识。而如果把当今全世界的信息平分给每一个活着的人，那么每个人拥有的信息量将足足超过当年亚历山大图书馆全部藏书的320倍。如果把所有这些信息刻到光盘上并且分五摞叠起来的话，那么这些光盘可以一直堆到月球。

这种数据爆炸是相对新鲜的现象。仅仅在2000年的时候，全世界全部的存储信息中还只有四分之一是数字化的，其余的都保存在纸张、胶片和其他模拟介质上。但是由于数字数据数量的增长十分迅速——几乎每三年就翻一番，这种情形很快发生了逆转。如今，在所有存储信息中只有不到2%是非数字化的。

鉴于如此悬殊的比例，人们免不了在理解大数据的时候仅仅从数量上进行考虑。然而这将会产生误导。大数据的另一个特征是它能够用数据来表现世界的众多层面，而这些层面以往从来都没有被量化过——这种特征可以被称为“数据化”。例如，位置信息的数据化最早是由于经纬度的发明，而最近又有了GPS。当计算机对几个世纪内的书籍进行取样时，文字便成了被处理的数据。甚至连友谊和爱好也被数据化了——例如通过Facebook。

借助于廉价的电脑内存、高性能处理器、智能算法、聪明软件以及从基本统计学中借鉴来的数学知识，这样的一类数据正在被应用于难以置信的新用途中。这种新方法并不是试图“教会”计算机去从事驾驶或翻译这样的事情，而是要向计算机输入足够多的信息，从而使它们能够推断概率，例如交通指示绿灯亮、红灯不亮的概率，或者是在特定语境下“light”一词意为“光”而不是“轻”的概率。

以这种方式对大量数据加以利用，要求人们在三个方面彻底改变对数据的态度。第一是收集和使用大量数据，而不是像统计学家们在过去100多年里所做的那样，只满足于少量的数据或样本。第二是抛弃人们对有条理和纯净的数据的偏爱，转而接受杂乱无章——在越来越多的情形下，少许的不精确是可以容忍的。第三，在许多场合，人们需要放弃对事情原委的追究，而代之以对相关性的接纳。利用大数据，而不是试图弄懂发动机抛锚或药物副作用消失的确切原因，研究人员可以收集和分析大量有关此类事件的信息及一切相关素材，找出可能有助于预测未来事件发生的规律。大数据有助于回答是什么、而不是为什么的问题——通常有这样的回答就足够了。

互联网重塑了人类交流的方式。大数据则不同：它标志着社会处理信息方式的变化。随着时间的推移，大数据可能会改变人们思考世界的方式。随着人们利用越来越多的数据来理解事情和作出决定，人们很可能会发现生活的许多层面是随机的、而不是确定的。

0.2 从因果关系到相关性

人们看待数据的方式的两个变化——从局部变为全部以及从纯净变为凌乱——催生了第三个变化：从因果关系到相关性。这代表着告别总是试图了解世界运转方式背后深层原因的态度，而走向仅仅需要弄清现象之间的联系以及利用这些信息来解决问题。

加拿大的研究人员正在开发一种大数据手段，以便能在明显症状出现之前发现早产婴儿体内的感染。通过把包括心率、血压、呼吸和血氧水平等16种生命体征转化成每秒1000多个数据点的信息流，他们已经能够找到极其轻微的变化与较为严重的问题之间的相关性。最终，这项技术将使医生能够提前采取行动，从而拯救生命。

大数据所产生的影响将远远超出医学和消费品的范畴：它将深远地改变政府的运作方式和政治的性质。在推动经济增长、提供公共服务或进行战争等方面，那些能够有效利用大数据的人将拥有胜过别人的巨大优势。迄今为止，最令人兴奋的成果出现在市级，在这个级别上获取数据和利用这些信息进行实验要容易一些。纽约市长迈克尔·布隆伯格（他本人就是靠着数据行业发家的）率先进行了一项努力：该市正在利用大数据改善公共服务和降低成本。其中一个例子就是新的火灾预防策略。

非法在屋内打隔断的建筑物着火的可能性比其他建筑物高很多。纽约市每年接到2.5万宗有关房屋住得过于拥挤的投诉，但市里只有200名处理投诉的巡视员。市长办公室一个分析专家小组觉得大数据可以帮助解决这一需求与资源的落差。该小组建立了一个市内全部90万座建筑物的数据库，并在其中加入市里19个部门所收集到的数据：欠税扣押记录、水电使用异常、缴费拖欠、服务切断、救护车使用、当地犯罪率、鼠患投诉，诸如此类。接下来，他们将这一数据库与过去5年中按严重程度排列的建筑物着火记录进行比较，希望找出相关性。果然，建筑物类型和建造年份是与火灾相关的因素。不过，一个没怎么预料到的结果是，获得外砖墙施工许可的建筑物与较低的严重火灾发生率之间存在相关性。

利用所有这些数据，该小组建立了一个可以帮助他们确定哪些住房拥挤投诉需要紧急处理的系统。他们所记录的建筑物的各种特征数据都不是导致火灾的原因，但这些数据与火灾隐患的增加或降低存在相关性。这种知识被证明是极具价值的：过去房屋巡视员出现场时签发房屋腾空令的比例只有13%，在采用新办法之后，这个比例上升到了70%——效率大大提高了。

0.3 大数据的挑战和局限

大数据也正在帮助提高民主政府的透明度。一个建立在“开放数据”概念上的运动已经形成，其诉求超出了目前在发达民主国家已经十分常见的信息自由法。这一运动的支持者呼吁政府把手上浩如烟海的普通数据向公众开放。

与此同时，在政府推动使用大数据的同时，它们还需要保护公众免受不正当市场垄断的侵害。管理大数据的法规甚至可能成为国家间的角斗场。出于对反托拉斯和保护隐私的关切，欧洲各国政府已经在严查谷歌公司。脸谱网可能会成为世界各地类似行动的打击目标，因为它持有太多的个人数据。外交官们应该准备好围绕是否像对待自由贸易那样对待信息流动展开交锋。

大数据势必将改变人们生活、工作和思考的方式。建立在强调因果关系基础上的世界观正在受到推崇相关性的挑战。知识的占有曾经意味着对历史的了解，而现在却意味着预言未来的能力。解决大数据所带来的挑战将不是易事。

在决策越来越多地受到数据支配的世界里，人、直觉或是不顾事实的蛮干还有什么用武之地呢？如果每个人都求助于数据，都利用大数据工具的话，那么不可预测性——例如人类的本能、冒险、意外甚至失误——也许将会成为差异的关键。如果真是这样的话，那么需要专门为人为因素辟出一席之地——即为直觉、常识、运气留出空间，以确保它们不会被数据和机器生成的答案挤走。

这将对社会进步的观念产生重要影响。大数据使我们可以更快地进行实验，对更多的线索展开探索。这些优势应该会导致更多创新的产生。但在有些时候，发明的火花迸发是数据所无法表现的。倘若亨利·福特当初求助于大数据算法系统来研究顾客希望得到的东西，算法系统得到的答案会是“更快的马匹”，也就不会有福特著名的汽车生产线了。在大数据的世界里，需要培养的恰恰是与人类关系最密切的特性——创造力、直觉和上进心，因为人的聪明才智才是进步的源泉。

大数据是一种资源和一种工具。它的目的是告知，而不是解释；它意在促进理解，但仍然会导致误解——关键在于人们对它的掌握程度。人们必须以一种不仅欣赏其力量，而且承认其局限的态度来接纳这种技术。

肯尼思·内尔·丘基尔是《经济学人》杂志数据编辑，维克托·梅耶—舍恩伯格是牛津大学互联网研究院互联网管理与法规教授。

1 函数

1.1 函数的定义

1.1.1 BuiltInFunctionDefinition​

属于Table API/SQL模块的​​函数定义层​​，是Flink对函数的统一抽象表示。
它定义了函数的基本属性（如名称、类型、参数策略等），与具体的执行引擎解耦，适用于所有API层（包括DataStream API和Table API）。

​​作用​​：

• 提供函数的元信息（如输入/输出类型策略、确定性等）
• 支持跨API的统一函数注册（如通过TableEnvironment.registerFunction()）
• 作为SQL和Table API函数调用的中间表示

1.1.2 FlinkSqlOperatorTable​

Operator table that contains only Flink-specific functions and operators.

是​​SQL解析层​​的组件，继承自Apache Calcite的SqlOperatorTable，专门用于SQL语法解析和验证阶段。
它定义了SQL标准中的运算符和函数（如+, SUM, OVER等）如何映射到Flink的执行逻辑。

​​作用​​：

• 提供SQL语法树（AST）中运算符和函数的符号表
• 在SQL解析阶段验证函数是否存在及参数合法性
• 将SQL函数名映射到BuiltInFunctionDefinition或其他执行逻辑

1.1.3 协同工作流程​

1. ​​SQL解析阶段​​：当用户提交SQL查询时，FlinkSqlOperatorTable负责识别函数名（如COUNT）并验证语法合法性
2. ​​逻辑计划生成​​：解析通过后，SQL函数会被转换为BuiltInFunctionDefinition实例，作为逻辑计划的一部分
3. ​​执行阶段​​：BuiltInFunctionDefinition提供具体的类型推导和执行逻辑，最终生成Flink作业图（JobGraph）

1.2 函数的校验

以IP_TO_LONG函数为例：

1
2
3
4
5
6
7
8
9

public static final SqlFunction IP_TO_LONG =  
        new SqlFunction(  
                "IpToLong",  // name
                SqlKind.OTHER_FUNCTION,  // SqlKind
                ReturnTypes.BIGINT_NULLABLE,  // SqlReturnTypeInference
                null,  // SqlOperandTypeInference
                OperandTypes.STRING,  // SqlOperandTypeChecker
                SqlFunctionCategory.STRING // SqlFunctionCategory
                );

1
2
3
4
5
6
7

// org.apache.calcite.sql.SqlOperator#checkOperandTypes
// org.apache.calcite.sql.type.FamilyOperandTypeChecker#checkOperandTypes
// org.apache.calcite.sql.type.FamilyOperandTypeChecker#checkSingleOperandType

...
family.getTypeNames().contains(typeName)
...

1.3 查找所有的函数定义

1
2
3
4
5
6
7
8
9

List<SqlOperator> operatorList = FlinkSqlOperatorTable.instance().getOperatorList();  
  
for (int i = 0; i < operatorList.size(); i++) {  
    SqlOperator sqlOperator = operatorList.get(i);  
    System.out.printf("FlinkSqlOperateTable: %03d,\t%-35s,%6s,\n",  
            i + 1,  
            sqlOperator.getName(),  
            sqlOperator.kind);  
}

转载原文

问题一： session模式，哪些组件是共用的？
问题二：每种模式的优缺点
问题三： runtime有哪些组件构成？
问题四: JobManager有哪些组件？

runtime summary

https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=65147077

JobManager

Dispatcher

提供Client的RPC接口

ResourceManager

与Yarn resource manager不同，用来管理Flink的TaskManager资源

JobMaster

一个任务的生命周期，每个Flink job各有一个JobMaster与之对应

Web server

TaskManager

TaskManager 其实功能比较简单，提供一些基本的 RPC 服务，供 JobManager 进行调度

集群启动

per-job mode

yarn session mode

application mode

参考文献

• flink on yarn 远程 debug 提交流程源码分析看不懂？看完这53张图包你学会！
  • Flink on yarn per-job的架构、提交流程和代码分析
  • 如何开启远程调试，本地调试
• Flink是如何工作的?
  • Flink启动顺序，各个组件启动流程
• 详解Flink作业提交流程
  • Session模式、Per-job模式、K8s session模式提交流程
• 剖析Flink运行时的整体架构
  • 各个组件的作用
• Apache Flink on Kubernetes: 四种模式，我该选择哪种？
• Apache Flink进阶（一）：Runtime核心机制

1 概述

1.1 Flink的优势

Apache Flink优势(Apache Flink Advantages)

1.2 Flink OLAP的优势

Flink做OLAP的优势

Apache Flink做OLAP引擎优势(The advantages of Apache Flink as an OLAP engine)

统一引擎(Unified Framework)

• 降低学习成本(Easy to learn)
• 提高开发效率(Improve development efficiency)
• 提高维护效率(Improve maintenance effciency)

既有优势(Existing strengths)

• 内存计算(In-Memory computing)
• Pipeline
• 代码动态生产(Code-gen)
• …

相互增强(Mutual reinforcement)

• 无统计信息的优化(Optimization without statistic)
• 更高效的算子(More efficient operators)
• 更通用的框架(More general framework)

1.3 Flink vs Spark

1.4 spark accumulate

用法

1
2
3
4
5
6
7
8

// 1：创建累加器
private IntCounter numLines = new IntCounter();
// 2：注册累加器
getRuntimeContext().addAccumulator("num-lines", this.numLines);
// 3：使用累加器
this.numLines.add(1);
// 4：获取累加器的结果
myJobExecutionResult.getAccumulatorResult("num-lines")

2 环境构建

TDFlink 本地调试

3 Runtime

3.1 push pull 模型?

barrier 对齐

3.2 watermark

watermark

kafka 的每个分区 event-time 到来的时间不一致

using Flink’s Kafka-partition-aware watermark generation. watermarks are generated inside the Kafka consumer, per Kafka partition, and the per-partition watermarks are merged in the same way as watermarks are merged on stream shuffles.

watermark 在每个 kafka 分区的 kafka 消费者中产生，每个分区的 watermark 合并起来，如在 stream shuffle 中合并 watermark 一样

3.3 时间语义:EventTime & ProcessTime & IngestionTime

Event time programs must specify how to generate Event Time Watermarks, which is the mechanism that signals progress in event time

Internally, ingestion time is treated much like event time, but with automatic timestamp assignment and automatic watermark generation.

3.4 算子之间数据传递，是如何序列化的

avro 与 kryo 序列化的区别, 及使用场景

• 2PC

• 分布式快照

3.5 对象重用Object-Reuse

#Flink优化 #对象重用 : https://www.jianshu.com/p/f0eaaf2eaacc

对象重用： chain起来的算子，无需进行对象的序列化和反序列化，节省计算资源。
然而，存在对象可变的问题，需要避免。

Flink SQL是较适合的场景，对象不可变，SQL的类型都是复杂类型，重用会节省更多。

3.6 类型系统

• 类型系统

• runtime

  • task chain

  • sharing slot group

  • task slot

    • CoLocationGroup

    • SlotSharingGroup

    • Flink job submit

3.7 部署模式

• Deployment mode

  • Flink on yarn
  • Flink on K8s

3.8 Submit

• clients
• job submit

3.9 一致性语义: at least once & exactly once

从容错和消息处理的语义上(at least once, exactly once)，Flink 引入了 state 和 checkpoint。

3.10 Failover

3.10.1 Restart Strategies(重启策略)

默认重启策略通过 Flink-conf.yaml 指定

常用的重启策略
固定间隔 (Fixed delay)
失败率 (Failure rate)
无重启 (No restart)
如果没有启用 checkpointing，则使用无重启 (no restart) 策略。
如果启用了 checkpointing，但没有配置重启策略，则使用固定间隔 (fixed-delay) 策略，其中 Integer.MAX_VALUE 参数是尝试重启次数
重启策略可以在 Flink-conf.yaml 中配置，表示全局的配置。也可以在应用代码中动态指定，会覆盖全局配置

1
2
3
4
5
6
7
8
9

第一种：全局配置 Flink-conf.yaml
restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: 3
restart-strategy.fixed-delay.delay: 10 s
第二种：应用代码设置
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
  3, // 尝试重启的次数
  Time.of(10, TimeUnit.SECONDS) // 间隔
));

3.11 Accumulators & Counters

累加器与计数器

4 状态/Checkpoint/Savepoint

状态托管
operator state 算子状态
keyed state

barrier

• 状态管理
• checkpoint api
• TTL 实现原理
• Flink-on-yarn ✅ 2026-01-16
• checkpoint 与 savepoint
• checkpoint 诊断
• RocksDB 调优 https://cloud.tencent.com/developer/article/1592441
• 状态与恢复
• backpress 反压、oppo 分享
• time
• Flink 内存管理
• 分布式缓存
• Flink HA

4.1 状态

默认保存在 java 堆中
checkpoint: 把 state 持久化存储，全局快照
blink 做的优化

失败恢复机制

状态分为原始状态和托管状态

托管状态是由 Flink 框架管理的状态
而原始状态，由用户自行管理状态具体的数据结构，框架在做 checkpoint 的时候，使用 byte[]来读写状态内容，对其内部数据结构一无所知。
通常在 DataStream 上的状态推荐使用托管的状态，当实现一个用户自定义的 operator 时，会使用到原始状态。

ReducingState 如何使用

4.2 Checkpoint

Flink 的 checkpoint 机制可以与(stream 和 state)的持久化存储交互的前提：
持久化的 source，它需要支持在一定时间内重放事件。这种 sources 的典型例子是持久化的消息队列（比如 Apache Kafka，RabbitMQ 等）或文件系统（比如 HDFS，S3，GFS 等）
用于 state 的持久化存储，例如分布式文件系统（比如 HDFS，S3，GFS 等）

Checkpoint 是 Flink 实现容错机制最核心的功能，它能够根据配置周期性地基于 Stream 中各个 Operator/task 的状态来生成快照，从而将这些状态数据定期持久化存储下来，当 Flink 程序一旦意外崩溃时，重新运行程序时可以有选择地从这些快照进行恢复，从而修正因为故障带来的程序数据异常

checkpoint 开启之后，默认的 checkPointMode 是 Exactly-once
checkpoint 的 checkPointMode 有两种，Exactly-once 和 At-least-once
Exactly-once 对于大多数应用来说是最合适的。At-least-once 可能用在某些延迟超低的应用程序（始终延迟为几毫秒）

ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:表示一旦 Flink 处理程序被 cancel 后，会保留 Checkpoint 数据，以便根据实际需要恢复到指定的 Checkpoint
ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: 表示一旦 Flink 处理程序被 cancel 后，会删除 Checkpoint 数据，只有 job 执行失败的时候才会保存 checkpoint

4.3 State backend

默认情况下，state 会保存在 taskmanager 的内存中，checkpoint 会存储在 JobManager 的内存中。
state 的 store 和 checkpoint 的位置取决于 State Backend 的配置
env.setStateBackend(…)
一共有三种 State Backend
MemoryStateBackend
FsStateBackend
RocksDBStateBackend

MemoryStateBackend
state 数据保存在 java 堆内存中，执行 checkpoint 的时候，会把 state 的快照数据保存到 jobmanager 的内存中
基于内存的 state backend 在生产环境下不建议使用
FsStateBackend
state 数据保存在 taskmanager 的内存中，执行 checkpoint 的时候，会把 state 的快照数据保存到配置的文件系统中
可以使用 hdfs 等分布式文件系统
RocksDBStateBackend
RocksDB 跟上面的都略有不同，它会在本地文件系统中维护状态，state 会直接写入本地 rocksdb 中。同时它需要配置一个远端的 filesystem uri（一般是 HDFS），在做 checkpoint 的时候，会把本地的数据直接复制到 filesystem 中。fail over 的时候从 filesystem 中恢复到本地
RocksDB 克服了 state 受内存限制的缺点，同时又能够持久化到远端文件系统中，比较适合在生产中使用

修改 State Backend 的两种方式
第一种：单任务调整
修改当前任务代码
env.setStateBackend(new FsStateBackend(“hdfs://namenode:9000/Flink/checkpoints”));
或者 new MemoryStateBackend()
或者 new RocksDBStateBackend(filebackend, true);【需要添加第三方依赖】
第二种：全局调整
修改 Flink-conf.yaml
state.backend: filesystem
state.checkpoints.dir: hdfs://namenode:9000/Flink/checkpoints
注意：state.backend 的值可以是下面几种：jobmanager(MemoryStateBackend), filesystem(FsStateBackend), rocksdb(RocksDBStateBackend)

5 Stream API

timer 使用与原理

5.1 stateful stream processing

processFunction

低阶 API
构建一些新的组件
比如 利用定时做一定情况下的匹配和缓存
灵活，开发比较复杂

如: 凌晨更新 注册定时器

5.2 allGrouping 元素广播

Flink broadcast how to use

broadcast vaiable value can not be modified

用法

1
2
3
4
5
6

1：初始化数据
DataSet<Integer> toBroadcast = env.fromElements(1, 2, 3)
2：广播数据
.withBroadcastSet(toBroadcast, "broadcastSetName");
3：获取数据
Collection<Integer> broadcastSet = getRuntimeContext().getBroadcastVariable("broadcastSetName");

• 广播变量不能太大
• 不能修改，保证数据一致性

5.3 Distributed Cache

Spark 分布式缓存如何使用

Flink 提供了一个分布式缓存，类似于 hadoop，可以使用户在并行函数中很方便的读取本地文件
此缓存的工作机制如下：程序注册一个文件或者目录(本地或者远程文件系统，例如 hdfs 或者 s3)，通过 ExecutionEnvironment 注册缓存文件并为它起一个名称。当程序执行，Flink 自动将文件或者目录复制到所有 taskmanager 节点的本地文件系统，用户可以通过这个指定的名称查找文件或者目录，然后从 taskmanager 节点的本地文件系统访问它
用法
1：注册一个文件
env.registerCachedFile(“hdfs:///path/to/your/file”, “hdfsFile”)
2：访问数据
File myFile = getRuntimeContext().getDistributedCache().getFile(“hdfsFile”);

5.4 Reduce 在流计算的应用

org.apache.Flink.streaming.api.functions.timestamps.AscendingTimestampExtractor$LoggingHandler:140 -
Timestamp monotony violated: 1568772496000 < 1568772500000

• Flink 三种 Sink 模式: Append、Upsert 和 Retract
• Flink AsyncIO
• Flink plan本地编译获取plan
• Stream API
• Flink Stream API
• Flink join: broadcast join、interval join、window join、cogroup

5.5 window

• window api
• 自定义 trigger、evictor
• session-window

5.5.1 自定义 Window

1、Window Assigner

负责将元素分配到不同的 window。

Window API 提供了自定义的 WindowAssigner 接口，我们可以实现 WindowAssigner 的

1

public abstract Collection<W> assignWindows(T element, long timestamp)

方法。同时，对于基于 Count 的 window 而言，默认采用了 GlobalWindow 的 window assigner，例如：

1

keyBy.window(GlobalWindows.create())

2、Trigger

Trigger 即触发器，定义何时或什么情况下移除 window

我们可以指定触发器来覆盖 WindowAssigner 提供的默认触发器。 请注意，指定的触发器不会添加其他触发条件，但会替换当前触发器。

3、Evictor（可选）

驱逐者，即保留上一 window 留下的某些元素

4、通过 apply WindowFunction 来返回 DataStream 类型数据。

利用 Flink 的内部窗口机制和 DataStream API 可以实现自定义的窗口逻辑，例如 session window。

6 Table API

• Flink join

6.1 over window

• Flink Table API office
• 自定义窗口
• cogroup、connect 与 join
• CDC
• Flink 流测试工具
• 今日活动总申购次数：并发度为 1，是否 windowAll 都会是并发度为 1，是否可以先 keyBy 再合并？

7 Flink SQL

• 语法糖
• 实践
• submit
• Catalog
• Flink SQL join
• Flink interval join
• Flink temporal join
• Flink broadcast join
• Flink SQL join 的状态管理
• Flink window & demo
• Flink SQL-window aggregate 与 group aggregate
• Flink over window 与 TopN
• Flink SQL TableFactory
• Flink SQL 原理之 SQL 执行流程
• UDF
• build in function

Flink SQL 原理:

• Flink SQL 执行原理系列

  • Flink-SQL-解析执行主流程

• Calcite 原理

• Flink SQL 原理与 blink 优化

• Flink SQL 代码生成

• Flink SQL 执行原理

Blink:

• Blink SQL

7.1 SQL运行

7.1.1 SqlNode

7.1.2 validate

`SqlNode validated = flinkPlanner.validate(sqlNode);``

sqlNode.accept 访问者模式

7.1.3 Operation

org.apache.flink.table.operations.Operation

org.apache.flink.table.planner.calcite.FlinkPlannerImpl#rel

org.apache.calcite.sql2rel.SqlToRelConverter

7.1.4 Rel

org.apache.flink.table.planner.delegation.PlannerBase#translateToRel

7.1.5 optimize

7.1.6 ExecNodeGraph

7.1.7 Plan/Transformation

org.apache.flink.table.delegation.Planner#translate

7.1.8 execute

org.apache.flink.table.api.internal.TableEnvironmentImpl#executeInternal(org.apache.flink.table.operations.Operation)

8 Flink metric/Reporter

• (Flink 累加器和计数器:Accumulators & Counters)(Flink-metrics)

9 Flink CEP

• CEP

10 Flink 调度

10.1 Flink on Yarn

Flink on K8s

11 Flink常见优化与优化参数

11.1 SQL mini-batch

11.2 object-reuse

12 Flink Connectors

• kudu connector
• Connectors
• Hippo Connector
• Kudu Connector
• [Kafka Connector][Flink-connector-kafka]
• HBase
• 数据源重放
• Flink-sink-mode
• async api scala sink

12.1 HadoopOutputFormat

Flink hbase connector

13 Flink 运维

• [Flink UI][flink-ui]
• Yarn、JM、TM 配置与参数
• 常见异常与解决方案

LatencyMarker

Flink全链路延迟的测量方式

14 Flink 最佳实践

• interval join 实现 outjoin
• 双流 join，一条流延迟
• keyedState 扩容
• Flink-bloomfilter 近似去重
• Flink 状态动态更新
• Flink 数据倾斜
• 实时数据同步
• 动态配置
• 欺诈检测
• 定时输出
• 延迟触发/超时统计

14.1 亿级流量 秒级统计

AI、ML、DL、ANN、CNN 蛰伏数十年，就为这个大数据时代。

AI 正以前所未有的速度改变世界，影响着我们生活的方方面面，AI 走进人们的生活将成为未来的新常态。 而大数据和计算能力像水和氧气一样支撑着 AI，我们的数据从手机端、服务器日志

14.2 实时维表

15 开源项目

• Oceanus

16 源码解读

根据上面我们选择的核心模块，我们大概可以得知，整个Flink的核心实现包含了：

• Flink 基本组件和逻辑计划：介绍了 Flink 的基本组件、集群构建的过程、以及客户端逻辑计划的生成过程
• Flink 物理计划生成：介绍了 Flink JobManager 对逻辑计划的运行时抽象，运行时物理计划的生成和管理等
• Jobmanager 基本组件和TaskManager的基本组件
• Flink 算子的生命周期：介绍了 Flink 的算子从构建、生成、运行、及销毁的过程
• Flink 网络栈：介绍了 Flink 网络层的抽象，包括中间结果抽象、输入输出管理、BackPressure 技术、Netty 连接等
• Flink的水印和Checkpoint
• Flink-scheduler：介绍 Flink 的任务调度算法及负载均衡
• Flink对用户代码异常处理：介绍作业的代码异常后 Flink 的处理逻辑，从而更好的理解 Flink 是如何保证了 exactly-once 的计算语义
• Flink Table/SQL 执行流程、Flink和Hive的集成等

17 参考文献

1. Flink 官方文档翻译
2. LittleMagic - 简书

18 社区

18.1 FLIP

https://cwiki.apache.org/confluence/display/FLINK/FLIP-280%3A+Introduce+EXPLAIN+PLAN_ADVICE+to+provide+SQL+advice

The dataflow model:
a practical approach to balancing correctness, latency, and cost in massive-scale, unbounded, out-of-order data processing
数据流模型:在大规模、无界、无序的数据处理中，平衡正确性、延迟和成本的实用方法

https://research.google/pubs/pub43864/

The Dataflow Model 是 Google Research 于2015年发表的一篇流式处理领域的具有指导性意义的论文，它对数据集特征和相应的计算方式进行了归纳总结，并针对大规模/无边界/乱序数据集，提出一种可以平衡准确性/延迟/处理成本的数据模型。这篇论文的目的不在于解决目前流计算引擎无法解决的问题，而是提供一个灵活的通用数据模型，可以无缝地切合不同的应用场景。（来源于：时间与精神小屋的论文总结）

本论文是通过机翻+人翻结合一起的，里面包含大量的长句，如果纯人翻的话，完全啃下来有点难！

0.1 ABSTRACT

Unbounded, unordered, global-scale datasets are increasingly common in day-to-day business (e.g. Web logs, mobileusage statistics, and sensor networks). At the same time,consumers of these datasets have evolved sophisticated requirements, such as event-time ordering and windowing by features of the data themselves, in addition to an insatiable hunger for faster answers. Meanwhile, practicality dictates that one can never fully optimize along all dimensions of correctness, latency, and cost for these types of input. As a result, data processing practitioners are left with the quandary of how to reconcile the tensions between these seemingly competing propositions, often resulting in disparate implementations and systems.

无边界的、无序的、全球范围的数据集在日常商业中越来越普遍（例如，Web日志，移动设备使用情况统计信息和传感器网络）。 同时，这些数据集的消费者已经提出了更加复杂的需求，例如基于event-time（事件时间）的排序和数据特征本身的窗口聚合，以满足消费者对于快速消费数据的庞大需求。与此同时，从实用性的角度出发，对于以上提到的数据集，我们永远无法在准确（correctness），延迟（latency）和成本（cost）等所有维度上进行全面优化。 最后，数据处理人员需要在这些看似冲突的方面之间做出妥协与调和，而这些做法往往会产生不同的实现与框架。

We propose that a fundamental shift of approach is necessary to deal with these evolved requirements in modern data processing. We as a field must stop trying to groom unbounded datasets into finite pools of information that eventually become complete, and instead live and breathe under the assumption that we will never know if or when we have seen all of our data, only that new data will arrive, old data may be retracted, and the only way to make this problem tractable is via principled abstractions that allow the practitioner the choice of appropriate tradeoffs along the axes of interest: correctness, latency, and cost.

我们认为有关于数据处理的方法必须得到根本性的改变，以应对现代数据处理中这些不断发展的需求。作为流式处理的领域中，我们必须停止尝试将无边界的数据集归整成完整的、有限的信息池，因为在一般的情况下，我们永远不知道是否或者何时能看到所有的数据。使得该问题变得易于解决的唯一方法就是通过一些规则上的抽象，使得数据处理人员能够从准确（correctness），延迟（latency）和成本（cost）几个维度做出妥协。

In this paper, we present one such approach, the Dataflow Model, along with a detailed examination of the semantics it enables, an overview of the core principles that guided its design, and a validation of the model itself via the real-world experiences that led to its development.

在本文中，我们提出了一种这样的方法，The Dataflow Model，并对其支持的语义进行了详细的审视，概述其设计指导的核心原则，并通过实际的开发经验验证了模型本身的可行性。

0.2 INTRODUCTION

Modern data processing is a complex and exciting field. From the scale enabled by MapReduce and its successors(e.g Hadoop, Pig, Hive, Spark), to the vast body of work on streaming within the SQL community (e.g.query systems, windowing, data streams,time domains, semantic models), to the more recent forays in low-latency processing such as Spark Streaming, MillWheel, and Storm, modern consumers of data wield remarkable amounts of power in shaping and taming massive-scale disorder into organized structures with far greater value. Yet, existing models and systems still fall short in a number of common use cases.

现代数据处理是一个复杂且令人兴奋的领域。从MapReduce及其继承者（e.g. Hadoop，Pig，Hive，Spark）实现的大规模运算，到SQL社区对流式处理做出的巨大工作（e.g. 查询系统（query system），窗口（windowing），数据流（data streams），时间域（time domains），语义模型（semantic model）），再到近期如Spark Streaming，MillWheel和Storm对于低延迟数据处理的初步尝试，现代数据的消费者挥舞着庞大的力量，尝试将大规模的、无序的海量数据规整为具有巨大价值的、易于管理的结构当中。然而，现有的模型和系统在许多常见的用例仍然存在不足的地方。

Consider an initial example: a streaming video provider wants to monetize their content by displaying video ads and billing advertisers for the amount of advertising watched. The platform supports online and offline views for content and ads. The video provider wants to know how much to bill each advertiser each day, as well as aggregate statistics about the videos and ads. In addition, they want to efficiently run offline experiments over large swaths of historical data.

考虑一个比较简单的例子：流视频提供者希望通过展示视频广告来使其视频内容能够盈利，并且通过广告的观看量对广告商收取一定的费用。该平台同时支持在线和离线观看视频和广告。视频提供者想要知道每天应向每个广告商收取多少费用，以及所有视频和广告的统计情况。此外，他们还希望能够有效率地对大量的历史数据进行离线实验。

Advertisers/content providers want to know how often and for how long their videos are being watched, with which content/ads, and by which demographic groups. They also want to know how much they are being charged/paid. They want all of this information as quickly as possible, so that they can adjust budgets and bids, change targeting, tweak campaigns, and plan future directions in as close to realtime as possible. Since money is involved, correctness is paramount.

而广告商/内容提供商想要知道他们的视频被观看的频率和时长，观看的内容/广告是什么，观看的人群是什么。他们也想知道他们需要为此要付出多少费用。他们希望尽可能快地获得所有这些信息，这样他们就可以调整预算和投标，改变目标，调整活动，并尽可能实时地规划未来的方向。因为涉及到钱，所以系统上设计时需要首要重点考虑其准确性。

Though data processing systems are complex by nature,the video provider wants a programming model that is simple and flexible. And finally, since the Internet has so greatly expanded the reach of any business that can be parceled along its backbone, they also require a system that can handle the diaspora of global scale data.

虽然数据处理系统本质上是复杂的，但是视频提供商却想要一个简单而灵活的编程模型。最后，由于互联网极大地扩展了任何可以沿着其主干分布的业务的范围，他们还需要一个能够处理全球范围内所有分散数据的系统。

The information that must be calculated for such a usecase is essentially the time and length of each video viewing,who viewed it, and with which ad or content it was paired(i.e. per-user, per-video viewing sessions). Conceptually this is straightforward, yet existing models and systems all fall short of meeting the stated requirements.

对于这样的一个用例，必须计算的信息本质上等同于每个视频观看的时长、谁观看了它，以及它与哪个广告或内容配对(e.g. 每个用户，每个视频观看会话)。从概念上讲，这很简单，但是现有的模型和系统都不能满足上述提到的需求。

Batch systems such as MapReduce (and its Hadoop vari-ants, including Pig and Hive), FlumeJava, and Spark suffer from the latency problems inherent with collecting all input data into a batch before processing it. For many streaming systems, it is unclear how they would remain fault-tolerantat scale (Aurora, TelegraphCQ, Niagara, Esper). Those that provide scalability and fault-tolerance fall short on expressiveness or correctness vectors.

诸如MapReduce（及其Hadoop变体，包括Pig和Hive），FlumeJava和Spark之类的批处理系统都碰到了在批处理之前需要将所有输入数据导入系统时所带来的延迟问题。对于许多流系统，我们无法清晰地知道他们是如何构建大规模的容错机制（Aurora，TelegraphCQ，Niagara，Esper），而那些提供可伸缩性和容错性的系统则在表达性或准确性方面上表现不足。

Many lack the ability to provide exactly-once semantics (Storm, Samza, Pulsar), impacting correctness. Others simply lack the temporal primitives necessary for windowing(Tigon), or provide windowing semantics that are limited to tuple- or processing-time-based windows (Spark Streaming, Sonora, Trident).

许多框架都缺乏提供exactly-once语义的能力（Storm，Samza，Pulsar），从而影响了数据的准确性。 而其他框架则缺少窗口所必需的时间原语（Tigon），或提供仅限于以元组（tuple-）或处理时间（processing-time）为基础的窗口语义（Spark Streaming，Sonora，Trident）。

Most that provide event-time-based windowing either rely on ordering (SQLStream),or have limited window triggering semantics in event-time mode (Stratosphere/Flink). CEDR and Trill are note worthy in that they not only provide useful triggering semantics via punctuations, but also provide an overall incremental model that is quite similar to the one we propose here; however, their windowing semantics are insufficient to express sessions, and their periodic punctuations are insufficient for some of the use cases in Section3.3. MillWheel and Spark Streaming are both sufficiently scalable, fault-tolerant, and low-latency to act as reasonable substrates, but lack high-level programming models that make calculating event-time sessions straightforward.

大多数框架提供的基于 event-time 的窗口机制要么依赖于排序（SQLStream），要么在event-time 模式下提供有限的窗口触发语义（Stratosphere / Flink）。值得一提的是，CEDR 和 Trill不仅可以通过标点符号（punctuations）提供有效的窗口触发语义，而且还提供了一个整体增量（overall incremental）的模型，该模型与我们此处提到的模型非常相似。然而，它们的窗口语义并不足以表达会话（sessions），并且它们的周期性标点符号不足以满足3.3节中的某些用例。MillWhell 和 Spark Streaming 都具有伸缩性，容错性和低延迟性，作为流框架合理的基础架构，但是其缺少能够让基于 event-time 的会话计算变得通俗易懂的高级编程模型。

The only scalable system we are aware of that supports a high-level notion of unaligned windows such as sessions is Pulsar, but that system fails to provide correctness, as noted above. Lambda Architecture systems can achieve many of the desired requirements, but fail on the simplicity axis on account of having to build and maintain two systems. Summingbird ameliorates this implementation complexity by abstracting the underlying batch and streaming systems behind a single interface, but in doing so imposes limitations on the types of computation that can be performed, and still requires double the operational complexity.

我们观察到唯一具有伸缩性，并且支持未对齐窗口（例如会话）这种高级概念的流数据系统是Pulsar，但是如上所述，该系统无法提供准确性。Lambda架构体系可以满足许多我们期望的要求，但是由于必须构建和维护两套系统，因此其在简单性这一维度上就注定失败。Summingbird通过在单一接口背后抽象底层的批系统和流系统，来改善其实现的复杂性，但是这样做会限制其可以执行的计算类型，并且仍会有两倍的操作复杂性。

None of these short comings are intractable, and systems in active development will likely overcome them in due time. But we believe a major shortcoming of all the models and systems mentioned above (with exception given to CEDR and Trill), is that they focus on input data (unbounded orotherwise) as something which will at some point become complete. We believe this approach is fundamentally flawed when the realities of today’s enormous, highly disordered datasets clash with the semantics and timeliness demanded by consumers. We also believe that any approach that is to have broad practical value across such a diverse and variedset of use cases as those that exist today (not to mention those lingering on the horizon) must provide simple, but powerful, tools for balancing the amount of correctness, latency, and cost appropriate for the specific use case at hand.

这些缺点都不是很难解决的，积极开发中的系统很可能会在适当的时候攻克它们。 但是我们认为，上述所有模型和系统（CEDR和Trill除外）的一个主要缺点是，它们只专注于那些最终在某些时刻达到完整的输入数据（无界或其他）。 我们认为，当现今庞大且高度混乱的数据集与消费者要求的语义和及时性发生冲突时，这种方法从根本上是有缺陷的。 我们还认为，任何在如今多样的用例中都具有广泛实用价值的方法（更不用说那些长期存在的用例）必须提供简单但强大的工具来平衡准确性，低延迟性和适合于特定用例的成本。

Lastly, we believe it is time to move beyond the prevailing mindset of an execution engine dictating system semantics; properly designed and built batch, micro-batch, and streaming systems can all provide equal levels of correctness, and all three see widespread use in unbounded data processing today. Abstracted away beneath a model of sufficient generality and flexibility, we believe the choice of execution engine can become one based solely on the practical underlying differences between them: those of latency and resource cost.

最后，我们认为是时候超越执行引擎决定系统语义的主流思维了。 经过正确设计和构建的批处理，微批处理和流传输系统都可以提供同等程度的准确性，并且这三者在当今的无边界数据处理中都可以得到了广泛使用。 在具有足够通用性和灵活性的模型下进行抽象，我们认为执行引擎的选择可以仅基于它们之间的实际潜在差异（即延迟和资源成本）进行选择。

Taken from that perspective, the conceptual contribution of this paper is a single unified model which:

• Allows for the calculation of event-time ordered results, windowed by features of the data themselves, over an unbounded, unordered data source, with correctness, latency, and cost tunable across a broad spectrum of combinations.
• Decomposes pipeline implementation across four related dimensions, providing clarity, composability, andflexibility:
  • – What results are being computed.
  • – Where in event time they are being computed.
  • – When in processing time they are materialized.
  • – How earlier results relate to later refinements.
• Separates the logical notion of data processing from the underlying physical implementation, allowing the choice of batch, micro-batch, or streaming engine to become one of simply correctness, latency, and cost.

从这个角度来看，本文提出了一个单一且统一的模型概念，即：

• 允许计算event-time排序的结果，并根据数据本身的特征在无边界，无序的数据源上进行窗口化，其准确性，延迟和成本可在多种组合中调整。
• 分解四个跨维度相关的管道实现，以提供清晰性，可组合性和灵活性：
  • – What 正在计算什么结果。
  • – Where 在事件发生时，它们被计算在哪里。
  • – When 何时在prcoessing-time内实现。
  • – How 早期的结果如何与后来的改进相联系。
• 将数据处理的逻辑概念与底层物理实现分开，允许批处理，微批处理或流引擎的选择成为准确性，延迟和成本中的一种。

Concretely, this contribution is enabled by the following:

• A windowing model which supports unaligned event-time windows, and a simple API for their creation and use (Section 2.2).
• A triggering model that binds the output times of results to runtime characteristics of the pipeline, with a powerful and flexible declarative API for describing desired triggering semantics (Section 2.3).
• An incremental processing model that integrates retractions and updates into the windowing and triggering models described above (Section 2.3).
• Scalable implementations of the above atop the MillWheel streaming engine and the FlumeJava batch engine, with an external reimplementation for GoogleCloud Dataflow, including an open-source SDK that is runtime-agnostic (Section 3.1).
• A set of core principles that guided the design of this model (Section 3.2).
• Brief discussions of our real-world experiences with massive-scale, unbounded, out-of-order data processing at Google that motivated development of this model(Section 3.3).

具体来说，这一模型可由下面几个概念形成：

• 窗口模型（A windowing model）。支持未对齐的event-time窗口，以及提供易于创建和使用窗口 API 的模型(章节2.2)。
• 触发模型（A triggering model ）。将输出的时间结果与具有运行特性的管道进行绑定，并提供功能强大且灵活的声明性 API，用于描述所需的触发语义（章节2.3）。
• 增量处理模型（incremental processing model）。将数据回撤功能和数据更新功能集成到上述窗口和触发模型中（章节2.3）。
• 可扩展的实现（Scalable implementations）。在MillWheel流引擎和FlumeJava批处理引擎之上的可扩展实现以及对GoogleCloud Dataflow的外部重新实现，包括与运行时无关的开源SDK（章节3.1）。
• 核心原则（core principles）。用于指导该模型设计的一组核心原则(章节3.2)。
• 真实经验（ real-world experiences ）。简要讨论了我们在Google上使用大规模，无边界，无序数据处理的真实经验，这些经验推动了该模型的发展（章节3.3）。

It is lastly worth noting that there is nothing magical about this model. Things which are computationally impractical in existing strongly-consistent batch, micro-batch, streaming, or Lambda Architecture systems remain so, with the inherent constraints of CPU, RAM, and disk left steadfastly in place. What it does provide is a common framework that allows for the relatively simple expression of parallel computation in a way that is independent of the underlying execution engine, while also providing the ability to dial in precisely the amount of latency and correctness for any specific problem domain given the realities of the data and resources at hand. In that sense, it is a model aimed at ease of use in building practical, massive-scale data processing pipelines.

最后值得注意的是，这个模型没有什么神奇之处。在现有的强一致批处理、微批处理、流处理或Lambda体系结构系统中，那些不现实的东西依旧存在，CPU、RAM和 Disk的固有约束仍然稳定存在。它所提供的是一个通用的框架，该框架允许以独立于底层执行引擎的方式对并行计算进行相对简单的表达，同时还提供了在现有数据和资源下，为任何特定问题精确计算延迟和准确性的能力。从某种意义上说，它是一种旨在易于使用的模型，可用于构建实用的大规模数据处理管道。

0.2.1 Unbounded/Bounded vs Streaming/Batch

When describing infinite/finite data sets, we prefer the terms unbounded/bounded over streaming/batch, because the latter terms carry with them an implication of the use of a specific type of execution engine. In reality, unbounded datasets have been processed using repeated runs of batch systems since their conception, and well-designed streaming systems are perfectly capable of processing bounded data. From the perspective of the model, the distinction of streaming or batch is largely irrelevant, and we thus reserve those terms exclusively for describing runtime execution engines.

当描述无限/有限数据集时，我们首选“无界/有界”这一术语而不是“流/批处理”，因为后者会带来使用特定类型执行引擎的隐含含义。 实际上，自从无边界数据集的概念诞生以来，就已经使用批处理系统的重复运行对其进行了处理，而精心设计的流系统则完全能够处理有边界的数据。 从模型的角度来看，流或批处理的区别在很大程度上是无关紧要的，因此，我们保留了那些专门用于描述运行时执行引擎的术语。

0.2.2 Windowing

Windowing slices up a dataset into finite chunks for processing as a group. When dealing with unbounded data, windowing is required for some operations (to delineate finite boundaries in most forms of grouping: aggregation,outer joins, time-bounded operations, etc.), and unnecessary for others (filtering, mapping, inner joins, etc.). For bounded data, windowing is essentially optional, though still a semantically useful concept in many situations (e.g. back-filling large scale updates to portions of a previously computed unbounded data source).

窗口化（Windowing）将数据集切成有限的数据块，以作为一组进行处理。 处理无边界数据时，某些操作（在大多数分组形式中描绘有限边界：聚合，外部联接，有时间限制的操作等）需要窗口化，而其他操作（过滤，映射，内部联接等）则不需要。 对于有界数据，窗口化在本质上是可选的，尽管在许多情况下仍然是语义上十分有用的概念（例如，回填大规模数据更新到先前计算的无界数据源的某些部分中）。

Windowing is effectively always time based, while many systems support tuple-based windowing, this is essentially time-based windowing over a logical time domain where elements in order have successively increasing logical timestamps. Windows may be either aligned, i.e. applied across all the data for the window of time in question, or unaligned, i.e. applied across only specific subsets of the data (e.g. per key) for the given window of time. Figure 1 highlights three of the major types ofwindows encountered when dealing with unbounded data.

实际上，窗口化总是基于时间的，虽然许多系统支持基于元组的窗口，但这本质上还是基于时间的窗口，并在逻辑时间域上，元素按顺序依次增加逻辑时间戳。窗口可以是对齐的，即在时间窗口中应用所有数据，也可以是未对齐的，即在给定时间窗口中只应用数据的特定子集(例如，每个键值)。图1突出显示了在处理无界数据时遇到的三种主要windows类型。

Fixed windows (sometimes called tumbling windows) are defined by a static window size, e.g. hourly windows or daily windows. They are generally aligned, i.e. every window applies across all of the data for the corresponding period of time. For the sake of spreading window completion load evenly across time, they are sometimes unaligned by phase shifting the windows for each key by some random value.

固定窗口（有时称为翻滚窗口）。固定窗口由静态窗口大小定义，例如每小时一次或每天一次。 它们通常是对齐的，即每个窗口都在相应的时间段内应用于所有数据。为了使窗口完成时间均匀地分布在整个时间上，有时通过将每个键的窗口位移某个随机值来使它们不对齐。

Sliding windows are defined by a window size and slide period, e.g. hourly windows starting every minute. The period may be less than the size, which means the windows may overlap. Sliding windows are also typically aligned; even though the diagram is drawn to give a sense of sliding motion, all five windows would be applied to all three keys inthe diagram, not just Window 3. Fixed windows are really a special case of sliding windows where size equals period.

滑动窗口。滑动窗口由窗口大小和滑动周期定义，例如每分钟启动一次统计每小时的窗口。周期可能会小于窗口大小，这意味着窗口之间可能会发生重叠。 滑动窗口通常也会对齐，即使绘制该图给人提供一种滑动的感觉，所有五个窗口也将应用于该图中的所有三个键，而不仅仅是窗口3。固定窗口实际上是窗口大小等于滑动周期大小的滑动窗口的一种特殊情况。

Sessions are windows that capture some period of activity over a subset of the data, in this case per key. Typically they are defined by a timeout gap. Any events that occur within a span of time less than the timeout are grouped together as a session. Sessions are unaligned windows. For example, Window 2 applies to Key 1 only, Window 3 to Key2 only, and Windows 1 and 4 to Key 3 only.

**会话窗口。**会话是捕获数据子集（在此情况下为每个键值）的一段时间活动的窗口。 通常，它们由超时时间间隔定义的。 在小于超时的时间间隔范围内发生的任何事件都被归为一个会话。 会话是未对齐的窗口。 例如，窗口2仅适用于键1，窗口3仅适用于键2，窗口1和4仅适用于键3。

0.2.3 Time Domains

When processing data which relate to events in time, there are two inherent domains of time to consider. Though captured in various places across the literature (particularly time management and semantic models, but also windowing, out-of-order processing, punctuations, heartbeats, watermarks, frames), the detailed examples in section 2.3 will be easier to follow with the concepts clearly in mind. The two domains of interest are:

• Event Time, which is the time at which the event itself actually occurred, i.e. a record of system clock time (for whatever system generated the event) at the time of occurrence.
• Processing Time, which is the time at which an event is observed at any given point during processing within the pipeline, i.e. the current time according to the system clock. Note that we make no assumptions about clock synchronization within a distributed system.

在处理与时间事件相关的数据时，需要考虑两个固有的时间域。虽然在文献的不同地方都已经提到过（特别是时间管理和语义模型，但也有窗口，无序处理，标点（punctuations），心跳，水印（watermarks），帧（frame）），详细的例子在章节2.3中展示，其将有助于帮助我们在脑海中更加清晰地掌握它。以下两个时间领域我们所关心的是：

• **事件时间（Event Time）。**即事件本身实际发生的时间，即系统时钟时间（对于生成事件的任何系统）在事件发生时的记录。
• **处理时间 （Processing Time）。**这是在流水线内处理期间在任何给定点观察到事件的时间，即根据系统时钟的当前时间。 注意，我们不对分布式系统中的时钟同步做任何假设。

Event time for a given event essentially never changes,but processing time changes constantly for each event as it flows through the pipeline and time marches ever forward. This is an important distinction when it comes to robustly analyzing events in the context of when they occurred.

给定事件的事件时间在本质上是不会改变，但是处理时间会随着事件在管道中的流动而不断变化，时间会不断前进。这是一个重要的区别，当它在事件发生的背景下进行清晰地分析时。

During processing, the realities of the systems in use (communication delays, scheduling algorithms, time spent processing, pipeline serialization, etc.) result in an inherent and dynamically changing amount of skew between the two domains. Global progress metrics, such as punctuations or watermarks, provide a good way to visualize this skew. For our purposes, we’ll consider something like MillWheel’swa-termark, which is a lower bound (often heuristically established) on event times that have been processed by the pipeline. As we’ve made very clear above, notions of completeness are generally incompatible with correctness, so we won’t rely on watermarks as such. They do, however, provide a useful notion of when the system thinks it likely that all data up to a given point in event time have been observed,and thus find application in not only visualizing skew, but in monitoring overall system health and progress, as well as making decisions around progress that do not require complete accuracy, such as basic garbage collection policies.

在处理过程中，市面上所有系统都会因为某些原因（通信延迟，调度算法，处理所花费的时间，流水线序列化等）导致两个时间域之间存在固有的，动态变化的偏移量。 诸如标点（punctuations）或水印（watermarks）之类的全局进度指标提供了一种可视化这种偏移量的好方法。为了我们的目的，我们将考虑使用MillWheel的水印，这是管道已处理的事件时间的下限（通常是启发式确定的）。 正如我们在上面非常清楚地指出的那样，完整性的概念通常与准确性是不兼容，因此我们不会像这样依赖水印。 但是，它们确实提供了一个有用的概念，即系统可在所有的数据中，观察那些给定的事件时间节点上的数据，因此不仅可以用于可视化其偏移量，而且可以用于监视整个系统的运行状况和进度， 以及围绕整体进度做出不要求准确性的决策，例如基本的垃圾回收策略。

In an ideal world, time domain skew would always bezero; we would always be processing all events immediately as they happen. Reality is not so favorable, however, and often what we end up with looks more like Figure 2. Starting around 12:00, the watermark starts to skew more away from real time as the pipeline lags, diving back close to real time around 12:02, then lagging behind again noticeably by the time 12:03 rolls around. This dynamic variance in skew is very common in distributed data processing systems, and will play a big role in defining what functionality is necessary for providing correct, repeatable results.

在理想的世界中，时间域的偏移量将始终为零，即我们将始终在事件发生时立即处理所有事件。但是，现实情况并非如此，通常，我们最终得到的结果看起来更像图2。从12:00开始，随着管线的滞后，水印开始偏离实时更多，然后回到接近实时12:02，然后到12:03时，又明显落后了。 时间偏移量的动态差异在分布式数据处理系统中非常常见，并且在定义提供准确，可重复的结果所需的功能方面将发挥重要作用。

0.3 DATAFLOW MODEL

In this section, we will define the formal model for the system and explain why its semantics are general enough to subsume the standard batch, micro-batch, and streaming models, as well as the hybrid streaming and batch semantics of the Lambda Architecture. For code examples, we will usea simplified variant of the Dataflow Java SDK, which itself is an evolution of the FlumeJava API.

在本节中，我们将定义系统的正式模型，并解释为什么它的语义足够通用到可以包含标准批处理、微批处理和流模型，以及Lambda架构的混合流处理和批处理语义。对于代码示例，我们将使用Dataflow Java SDK的简化变体，它本身是FlumeJava API的演化。

0.3.1 Core Primitives

To begin with, let us consider primitives from the classic batch model. The Dataflow SDK has two core transforms that operate on the (key, value) pairs flowing through the system:

• ParDo for generic parallel processing. Each input element to be processed (which itself may be a finite collection) is provided to a user-defined function (called a DoFn in Dataflow), which can yield zero or more out-put elements per input. For example, consider an operation which expands all prefixes of the input key, duplicating the value across them:

  

• GroupByKey for key-grouping (key, value) pairs.

首先，让我们考虑经典批处理模型中的原语。Dataflow SDK有两个核心转换（transforms），它们对流经系统的（key、value）对进行操作：

• ParDo。ParDo用于通用并行处理。每个输入元素（它本身可能是一个有限的集合）均会被用户自定义的函数（在数据流中称为DoFn）所处理，该函数可以为每个输入生成零个或多个输出元素。例如，考虑这样一个操作，它展开输入key的所有前缀，在它们之间复制所有的value
• GroupByKey。GroupByKey用来基于 key键将数据进行聚合

The ParDo operation operates element-wise on each input element, and thus translates naturally to unbounded data.The GroupByKey operation, on the other hand, collects all data for a given key before sending them downstream for reduction. If the input source is unbounded, we have no way of knowing when it will end. The common solution to this problem is to window the data.

ParDo操作是在每个输入元素上逐个操作元素，从而能够很自然地将其转换为无界数据。而在另一方面，GroupByKey操作收集给定key键的所有数据，然后将它们发送到下游进行缩减。如果输入源是无界的，我们无法知道它何时结束。这个问题的常见解决方案是将数据窗口化。

0.3.2 Windowing

Systems which support grouping typically redefine their GroupByKey operation to essentially be GroupByKeyAndWindow. Our primary contribution here is support for un-aligned windows, for which there are two key insights. The first is that it is simpler to treat all windowing strategies as unaligned from the perspective of the model, and allow underlying implementations to apply optimizations relevant to the aligned cases where applicable. The second is that windowing can be broken apart into two related operations:

• Set<Window> AssignWindows(T datum), which assigns the element to zero or more windows. This is essentially the Bucket Operator from Li.

• Set<Window> MergeWindows(Set<Window> windows), which merges windows at grouping time. This allows data-driven windows to be constructed over time as data arrive and are grouped together.

支持分组的系统通常将GroupByKey操作重新定义为GroupByKeyAndWindow。我们在这里的主要贡献是支持未对齐的窗口，对此有两个关键的见解。首先，从模型的角度来看，将所有的窗口策略视为未对齐的比较简单，并允许底层实现对对齐的情况应用相关的优化。第二，窗口可以分解为两个相关的操作：

• Set<Window> AssignWindows(T datum)，它将元素赋值给零个或多个窗口。

• Set<Window> MergeWindows(Set<Window> windows)，它允许按时间分组时合并窗口。这允许在数据到达并分组在一起时，随时间构建数据驱动窗口。

For any given windowing strategy, the two operations are intimately related; sliding window assignment requires slid-ing window merging, sessions window assignment requires sessions window merging, etc.

对于任何给定的窗口策略，这两个操作都是密切相关的，如滑动窗口分配需要滑动窗口合并，会话窗口分配需要会话窗口合并，等等。

Note that, to support event-time windowing natively, instead of passing (key, value) pairs through the system, we now pass (key, value, eventtime, window) 4-tuples. Elements are provided to the system with event-time timestamps (which may also be modified at any point in the pipeline), and are initially assigned to a default global window, covering all of event time, providing semantics that match the defaults in the standard batch model.

注意，为了在本地支持事件时间的窗口，我们现在传递(key, value, eventtime, window) 4元组，而不是传递(key, value)到系统中。元素以基于事件时间的时间戳（也可以在管道中的任何位置修改）提供给系统，并在最初时分配给一个默认的全局窗口，覆盖所有事件时间，提供与标准批处理模型中的默认值匹配的语义。

0.3.2.1 Window Assignment

From the model’s perspective, window assignment creates a new copy of the element in each of the windows to which it has been assigned. For example, consider windowing a dataset by sliding windows of two-minute width and one-minute period, as shown in Figure 3 (for brevity, timestamps are given in HH:MM format).

从模型的的角度来看，窗口赋值是在每个已赋值给它的窗口中创建元素的新副本。例如，考虑使用两分钟时间长度和以一分钟为时间周期的滑动窗口来窗口化一个数据集，如图3所示。

In this case, each of the two (key, value) pairs is duplicated to exist in both of the windows that overlapped the element’s timestamp. Since windows are associated directly with the elements to which they belong, this means window assignment can happen any where in the pipeline before grouping is applied. This is important, as the grouping operation may be buried somewhere downstream inside a composite transformation (e.g.Sum.integersPerKey()).

在本例中，这两个(key, value)对中的每一个都被复制到重叠元素时间戳的两个窗口中。由于窗口直接与它们所属的元素相关联，这意味着在应用分组之前，可以在管道中的任何位置进行窗口分配。这很重要，因为分组操作可能隐藏在复合转换(例如Sum.integersPerKey())下游中的某个地方。

0.3.2.2 Window Merging

Window merging occurs as part of the GroupByKeyAndWindow operation, and is best explained in the context of an example. We will use session windowing since it is our motivating use case. Figure 4 shows four example data, three for k1 and one for k2, as they are windowed by session, with a 30-minute session timeout. All are initially placed in a default global window by the system. The sessions implementation of AssignWindows puts each element into a single window that extends 30 minutes beyond its own timestamp; this window denotes the range of time into which later events can fall if they are to be considered part of the same session. We then begin the GroupByKeyAndWindow operation, which is really a five-part composite operation:

• DropTimestamps - Drops element timestamps, as only the window is relevant from here on out.
• GroupByKey - Groups (value, window) tuples by key.
• MergeWindows - Merges the set of currently buffered windows for a key. The actual merge logic is defined by the windowing strategy. In this case, the windows for v1 and v4 overlap, so the sessions windowing strategy merges them into a single new, larger session, as indicated in bold.
• GroupAlsoByWindow - For each key, groups values by window. After merging in the prior step,v1 and v4 are now in identical windows, and thus are grouped together at this step.
• ExpandToElements - Expands per-key, per-window groups of values into (key, value, eventtime, window)tuples, with new per-window timestamps. In this example, we set the timestamp to the end of the window, but any timestamp greater than or equal to the timestamp of the earliest event in the window is valid with respect to watermark correctness.

窗口合并是GroupByKeyAndWindow操作的一部分，这将会在后面的示例中进行解释。我们因其常见性，决定在本例中使用会话窗口。图4显示了四个示例数据，其中三个用于k1，一个用于k2，因为它们是按会话窗口显示的，并且有30分钟的会话超时。它们最初都被系统放置在一个默认的全局窗口中。AssignWindows的会话实现将每个元素放入一个单独的窗口中，这个窗口比它自身的时间戳延长了30分钟。此窗口表示如果迟到的事件被认为是同一会话的一部分的话，它们可能落入的时间范围。然后我们开始GroupByKeyAndWindow操作，这实际上是一个由五部分组成的复合操作：

• DropTimestamps -丢弃元素时间戳，因为从这里开始，只有窗口相关的部分。

• GroupByKey -按key分组成(value、window)元组。

• MergeWindows -合并key的当前缓冲窗口集。实际的合并逻辑是由窗口策略定义的。在这种情况下，v1和v4对应的窗口重叠，所以会话窗口将它们合并成一个新的、更大的会话。

• GroupAlsoByWindow -对于每个key，通过窗口聚合所有的value。在前一步合并之后，v1和v4现在位于相同的窗口中，因此在这一步将它们组合在一起。

• ExpandToElements -将每个key、每个窗口的value组扩展为(key、value、eventtime、window)元组，并使用新的窗口时间戳。在本例中，我们将时间戳设置在窗口的末端，但任何大于或等于窗口中最早事件的时间戳的事件时间戳在水印准确性方面都被认为是有效的。

0.3.2.3 API

As a brief example of the use of windowing in practice,consider the following Cloud Dataflow SDK code to calculate keyed integer sums:

作为实际使用窗口的简要示例，请考虑以下Cloud Dataflow SDK代码以计算key 对应的整数和：

To do the same thing, but windowed into sessions with a 30-minute timeout as in Figure 4, one would add a single Window.into call before initiating the summation:

要执行相同的操作，但是要像图4那样以30分钟的超时时间窗口化到会话中，则需要在启动求和之前添加单个Window.into调用

0.3.3 Triggers & Incremental Processing

The ability to build un-aligned, event-time windows is an improvement, but now we have two more shortcomings to address:

• We need some way of providing support for tuple- and processing-time-based windows, otherwise we have regressed our windowing semantics relative to other systems in existence.
• We need some way of knowing when to emit the results for a window. Since the data are unordered with respect to event time, we require some other signal to tell us when the window is done.

拥有构建未对齐（un-aligned）的事件时间（event-time）窗口的能力是一种改进，但现在我们还有两个缺点需要解决：

• 我们需要某种方式来提供对基于元组和基于处理时间的窗口的支持，否则我们已经倒退了与现有的其他系统相关的窗口语义了。

• 我们需要一些方法知道什么时候发出窗口的结果。由于数据对于事件时间是无序的，我们需要一些其他信号来告诉我们什么时候窗口完成数据处理了。

The problem of tuple- and processing-time-based windows we will address in Section 2.4, once we have built up a solution to the window completeness problem. As to window completeness, an initial inclination for solving it might be to use some sort of global event-time progress metric, such as watermarks. However, watermarks themselves have two major shortcomings with respect to correctness:

• They are sometimes too fast, meaning there may be late data that arrives behind the watermark. For many distributed data sources, it is intractable to derive a completely perfect event time watermark, and thus impossible to rely on it solely if we want 100% correctness in our output data.
• They are sometimes too slow. Because they are a global progress metric, the watermark can be heldback for the entire pipeline by a single slow datum. And even for healthy pipelines with little variability in event-time skew, the baseline level of skew may still be multiple minutes or more, depending upon the input source. As a result, using watermarks as the sole signal for emitting window results is likely to yield higher latency of overall results than, for example, a comparable Lambda Architecture pipeline.

一旦我们建立了一个窗口完整性问题的解决方案，我们将在章节2.4中讨论基于元组和处理时间的窗口的问题。至于窗口完整性，解决它的最初倾向可能是使用某种全局的事件时间进度度量工具，例如水印（watermark）。但是，就准确性而言，水印（watermark）本身有两大缺点：

• 他们有时太快了，这意味着可能有迟来的数据可能会到达在水印后面。对于许多分布式数据源而言，它们很难获得十分完美的事件时间水印，因此如果我们想要输出数据100%正确，就不可能完全依赖于它。

• 他们有时太慢了。因为它们是一个全局进度度量，所以水印或许会被一个缓慢的数据来阻止整个管道。即使是在正常的管道中，即使在事件时间偏移量变化不大，偏移量的基线水平仍然可能是几分钟甚至更多，这取决于输入源。因此，使用水印作为唯一的信号来发送窗口结果可能会产生比类似的Lambda架构管道更高的延迟。

For these reasons, we postulate that watermarks alone are insufficient. A useful insight in addressing the completeness problem is that the Lambda Architecture effectively sidesteps the issue: it does not solve the completeness problem by somehow providing correct answers faster; it simply provides the best low-latency estimate of a result that the streaming pipeline can provide, with the promise of eventual consistency and correctness once the batch pipeline runs. If we want to do the same thing from within a single pipeline (regardless of execution engine), then we will need a way to provide multiple answers (or panes) for any given window.We call this feature triggers, since they allow the specification of when to trigger the output results for a given window.

由于这些原因，我们假定仅有水印（watermark）是不够的。解决窗口完整性问题的一个有用的方式（也是Lambda架构提出的一种有效回避该问题的方式）:它并没有更快地通过某种方式提供正确的解决方法来处理完整性问题，而只是提供了流管道所能提供的结果的最佳低延迟估计值，并承诺一旦批处理管道运行起来，将在最终保持一致性和正确性。如果我们希望在单个管道中执行相同的操作(与执行引擎无关)，那么我们将需要为任何给定窗口提供多个解决方法(或窗格)的方法。我们将此功能称为触发器（triggers），因为它们允许指定何时触发给定窗口的输出结果。

In a nutshell, triggers are a mechanism for stimulating the production of GroupByKeyAndWindow results in response to internal or external signals. They are complementary to the windowing model, in that they each affect system behaviour along a different axis of time:

• Windowing determines where in event time data are grouped together for processing.

• Triggering determines when in processing time the results of groupings are emitted as panes.

简而言之，触发器是一种机制，用于触发GroupByKeyAndWindow结果的生成，以响应内部或外部信号。它们是窗口模型的补充，因为它们都影响系统在不同时间轴上的行为：

• 窗口确定事件时间数据在哪里分组，并进行处理。

• 触发器决定在处理时间内分组的结果在什么时候以窗格的形式发出。

Our systems provide predefined trigger implementations for triggering at completion estimates (e.g. watermarks, including percentile watermarks, which provide useful semantics for dealing with stragglers in both batch and streaming execution engines when you care more about processing a minimum percentage of the input data quickly than processing every last piece of it), at points in processing time, and in response to data arriving (counts, bytes, data punctuations, pattern matching, etc.). We also support composing triggers into logical combinations (and, or, etc.), loops, sequences,and other such constructions. In addition, users may define their own triggers utilizing both the underlying primitives of the execution runtime (e.g. watermark timers, processing-time timers, data arrival, composition support) and any other relevant external signals (data injection requests, external progress metrics, RPC completion callbacks, etc.).We will look more closely at examples in Section 2.4.

我们的系统提供了用于在完成估算时触发的预定义触发器实现（例如，水印，包括百分位数水印，当您更关心快速处理最小百分比的输入数据而不是处理数据时，它们提供了有用的语义来处理批处理和流执行引擎中的散乱消息数据的最后一部分），当位于在处理时间点或者需要对数据到达（计数，字节，数据标点，模式匹配等）的响应时。 我们还支持将触发器组合成逻辑组合（and，or等），循环，序列和其他类似的构造。 另外，用户可以利用执行运行时的基本原语（例如水印计时器，处理时间计时器，数据到达，合成支持）和任何其他相关的外部信号（数据注入请求，外部进度指标，RPC回调等）来定义自己的触发器。。我们将在章节2.4中更详细地研究示例。

In addition to controlling when results are emitted, the triggers system provides a way to control how multiple panes for the same window relate to each other, via three different refinement modes:

• Discarding: Upon triggering, window contents are discarded, and later results bear no relation to previous results. This mode is useful in cases where the downstream consumer of the data (either internal or external to the pipeline) expects the values from various trigger fires to be independent (e.g. when injecting into a system that generates a sum of the values injected). It is also the most efficient in terms of amount of data buffered, though for associative and commutative operations which can be modeled as a Dataflow Combiner, the efficiency delta will often be minimal. For our video sessions use case, this is not sufficient, since it is impractical to require downstream consumers of our data to stitch together partial sessions.

• Accumulating: Upon triggering, window contents are left intact in persistent state, and later results become a refinement of previous results. This is useful when the downstream consumer expects to overwrite old values with new ones when receiving multiple results for the same window, and is effectively the mode used in Lambda Architecture systems, where the streaming pipeline produces low-latency results, which are then overwritten in the future by the results from the batch pipeline. For video sessions, this might be sufficient if we are simply calculating sessions and then immediately writing them to some output source that supports updates (e.g. a database or key/value store).

• Accumulating & Retracting: Upon triggering, inaddition to the Accumulating semantics, a copy of the emitted value is also stored in persistent state. When the window triggers again in the future, a retraction for the previous value will be emitted first, followed by the new value as a normal datum. Retractions are necessary in pipelines with multiple serial GroupByKeyAndWindow operations, since the multiple results generated by a single window over subsequent trigger fires may end up on separate keys when grouped downstream. In that case, the second grouping operation will generate incorrect results for those keys unless it is informed via a retraction that the effects of the original output should be reversed. Dataflow Combiner operations that are also reversible can support retractions efficiently via an uncombine method. For video sessions,this mode is the ideal. If we are performing aggregations downstream from session creation that depend on properties of the sessions themselves, for example detecting unpopular ads (such as those which are viewed for less than five seconds in a majority of sessions), initial results may be invalidated as inputs evolve overtime, e.g. as a significant number of offline mobile viewers come back online and upload session data. Retractions provide a way for us to adapt to these types of changes in complex pipelines with multiple serial grouping stages.

除了控制何时发出结果，触发器系统还提供了一种方法，可通过三种不同的优化模式来控制同一窗口的多个窗格之间的相互关系：

• 丢弃（Discarding）：触发器触发时，窗口内容将会被丢弃，并且以后的结果将与以前的结果无关。 倘若数据的下游使用者（管道内部或外部）期望来自各种触发器触发的值是独立的情况下（例如，注入到生成注入值之和的系统中），此模式很有用。 就缓冲的数据量而言，它也是最有效的，尽管对于可以为数据流组合器建模的关联和交换操作，增量效率通常会很小。 对于我们的视频会话用例，这是不够的，因为要求数据的下游使用者将部分会话缝合在一起是不切实际的。
• 累加（Accumulating）：触发器触发时，窗口内容将保持不变，以后的结果是以以前结果为基础，进行数据增量操作。这是十分有用的方法，当下游使用者希望在同一窗口中接收到多个结果时希望用新值覆盖旧值，并且系统能够有效地作用于Lambda架构系统。而在这其中，流管道产生低延迟的结果，这些结果随后将被来自批处理管道的结果覆盖。对于视频会话，如果我们只是简单地计算会话，然后立即将其写入支持更新的某个输出源中（例如数据库或key/value存储），这可能就足够了。
• 累积和回退（Accumulating & Retracting）：触发器触发时，除了累积语义外，输出值的副本也以持久状态存储。 当窗口在未来再次触发时，将首先会对先前值的回退，然后是输出作为正常基准的新值。 在具有多个串行GroupByKeyAndWindow操作的管道中，回退操作是必要的，因为在下游分组时，单个窗口在后续触发器触发上生成的多个结果可能会在单独的键上结束。 在那种情况下，第二次分组操作将为那些键生成不正确的结果，除非通过回退通知其原始输出进行回退。 数据流Combiner操作也可以通过取消组合方法有效地支持回退。 对于视频会话，此模式是理想的。 如果我们在会话创建的下游执行依赖于会话本身属性的聚合，例如检测不受欢迎的广告（例如在大多数会话中观看时间少于五秒钟的广告），则随着输入的发展，初始结果可能会是无效的，例如因为大量的离线移动设备恢复了在线状态并上传了会话数据。 回退为我们提供了一种方法，使我们可以通过多个串行分组阶段来适应复杂管道中的这些类型的更改。

0.3.4 Examples

举例部分比较简单，就是结合上面提到的所有概念，进行综合举例，有空再挖坑回填。

0.4 IMPLEMENTATION & DESING

实现部分是作者自身在 Google 内部的实践与经验，对于流系统开发者而言能够了解到他们在实现时碰到的坑。因为是了解背后原理就不进行详细翻译了。

0.5 CONCLUSIONS

The future of data processing is unbounded data. Though bounded data will always have an important and useful place, it is semantically subsumed by its unbounded counterpart. Furthermore, the proliferation of unbounded datasets across modern business is staggering. At the same time, consumers of processed data grow savvier by the day, demanding powerful constructs like event-time ordering and unaligned windows. The models and systems that exist today serve as an excellent foundation on which to build the data processing tools of tomorrow, but we firmly believe that a shift in overall mindset is necessary to enable those tools to comprehensively address the needs of consumers of unbounded data.

无边界（无限）的数据是数据处理的未来。 尽管有边界（有限）的数据将始终具有重要和有用的位置，但从语义上讲，它由无边界的对应部分所包含。 此外，无限数据集在整个跨现代业务中的扩散令人震惊。 同时，处理数据的消费者一天比一天更加精明，因此需要强大的架构，例如事件时间顺序和未对齐的窗口等。 当今存在的模型和系统为构建未来的数据处理工具奠定了良好的基础，但是我们坚信，必须转变整体的观念，以使这些工具能够全面满足数据消费者的需求。

Based on our many years of experience with real-world,massive-scale, unbounded data processing within Google, we believe the model presented here is a good step in that direction. It supports the un-aligned, event-time-ordered windows modern data consumers require. It provides flexible triggering and integrated accumulation and retraction, refocusing the approach from one of finding completeness in data to one of adapting to the ever present changes manifest in real-world datasets. It abstracts away the distinction of batch vs.micro-batch vs. streaming, allowing pipeline builders a more fluid choice between them, while shielding them from the system-specific constructs that inevitably creep into models targeted at a single underlying system. Its overall flexibility allows pipeline builders to appropriately balance the dimensions of correctness, latency, and cost to fit their use case, which is critical given the diversity of needs in existence. And lastly, it clarifies pipeline implementations by separating the notions of what results are being computed, where in event time they are being computed, when in processing time they are materialized, and how earlier results relate to later refinements. We hope others will find this model useful as we all continue to push forward the state of the art in this fascinating, remarkably complex field.

根据我们多年在Google中真实，大规模，无边界数据处理的经验，我们相信此处介绍的模型是朝这个方向迈出的重要一步。 它支持消费者需要的未对齐，事件时间顺序的窗口现代数据。 它提供了灵活的触发方式以及集成的累积和回退功能，将寻找数据完整性的方法重新定位为适应现实数据集中不断变化的方法。 它抽象化了批处理、微型批处理和流式处理三者的区别，使管道构建器可以在它们之间进行更多的选择，同时使它们免受系统特定的构造的影响，这些构造不可避免地会渗入针对单个基础系统的模型。 它的整体灵活性使流水线构建者可以适当地平衡正确性，延迟和成本这三个维度，以适应其用例，考虑到现有需求的多样性，这一点至关重要。最后，它通过分离以下概念来澄清流水线实现：正在计算哪些结果，其中计算它们的事件时间，在处理时间何时实现它们，以及较早的结果与以后的改进有何关系。我们希望其他人会发现此模型有用，因为我们所有人都将继续在这个引人入胜，非常复杂的领域中发展最先进的技术。

1. 现代流式计算的基石：Google DataFlow-阿里云开发者社区

2. 《The Dataflow Model》论文翻译

org.apache.flink.table.functions

所有的内置函数

org.apache.flink.table.functions.BuiltInFunctionDefinitions

以Date_Format为例

1
2
3
4
5
6
7
8
9

// 类型为SCALAR函数
// TypeStrategies: 推断函数调用的输出或累加器数据类型的策略。 
// 输出类型为空
public static final BuiltInFunctionDefinition DATE_FORMAT =
		new BuiltInFunctionDefinition.Builder()
			.name("dateFormat")
			.kind(SCALAR)
			.outputTypeStrategy(TypeStrategies.MISSING)
			.build();

hudi-indexing-mechanisms

Apache #Hudi 使用索引来定位更删操作所在的文件组。对于Copy-On-Write表，索引能加快更删的操作，因为避免了通过连接整个数据集来决定哪些文件需要重写。对于Merge-On-Read表，这个设计，对于任意给定的基文件，能限定要与其合并的记录数量。具体地，一个给定的基文件只需要和其所包含的记录的更新合并。相比之下，没有索引的设计（比如Apache Hive ACID），可能会导致需要把所有基文件与所有更删操作合并。

从高角度看，索引把一个记录的键加一个可选的分区路径映射到存储上的文件组ID（更多细节参考这里）。在写入过程中，我们查找这个映射然后把更删操作导向基文件附带的日志文件（MOR表），或者导向最新的需要被合并的基文件（COW表）。索引也使得Hudi可以根据记录键来规定一些唯一性的限制。

记录更新（黄色块）和基文件（白色块）合并消耗的对比

目前Hudi已经支持了几种不同的索引机制，并且在工具库中不断地完善和增加更多机制，在本文余下的篇幅中将根据我们的经验介绍几种不同作业场景下索引机制的选择。我们也会穿插一些对已有限制、即将开展的工作、以及优化和权衡方面的评论。

Hudi的索引类型

目前Hudi支持以下几种索引类型。

• 布隆索引（默认）：使用以记录的键生成的布隆过滤器，也可以用记录键对可能对应的文件进行剪枝操作。
• 简单索引：对更删的记录和存储上的表里提取的键进行轻量级的连接。
• HBase索引：使用外部的Apache HBase表来管理索引映射。

写入器可以通过hoodie.index.type来设置以上的类型。此外，自定义的索引实现可以通过hoodie.index.class来配置。对Apache Spark写入器需提供SparkHoodieIndex的子类。

另一个需要了解的关键点是区分全局索引和非全局索引。布隆索引和简单索引都有一个全局选项，分别是hoodie.index.type=GLOBAL_BLOOM和hoodie.index.type=GLOBAL_SIMPLE。HBase索引本质上就是全局索引。

• 全局索引：全局索引在全表的所有分区范围下强制要求键的唯一性，也就是确保对给定的键有且只有一个对应的记录。全局索引提供了更强的保证，而更删的消耗随着表的大小增加而增加（O(表的大小)），但仍可能对小表适用。
• 非全局索引：这个默认的索引实现只在一个分区里强制要求了这样的限制。由此可见，非全局索引依靠写入器为同一个记录的更删提供一致的分区路径，但由此同时大幅提高了效率，因为索引查询复杂度成了O(更删的记录数量)且可以很好地应对写入量的扩展。

由于数据可能有着不同的体量、速度和读取规律，不同索引会适用于不同的作业场景。接下来让我们分析几个不同的场景来讨论如何选择适合的Hudi索引。

作业场景：对事实表的延迟更新

许多公司会在NoSQL数据存储中存放大量的交易数据。例如共享出行的行程表、股票买卖记录的表、和电商的订单表。这些表通常一直在增长，且大部分的更新随机发生在较新的记录上，而对旧记录有着长尾分布型的更新。这通常是源于交易关闭或者数据更正的延迟性。换句话说，大部分更新会发生在最新的几个分区上而小部分会在旧的分区。

典型的事实表的更新样式。

对于这样的作业模式，布隆索引就能表现地很好，因为查询索引可以靠设置得当的布隆过滤器来剪枝很多数据文件。另外，如果生成的键可以以某种顺序排列，参与比较的文件数会进一步通过范围剪枝而减少。Hudi用所有文件的键域来构造区间树，这样能来高效地依据输入的更删记录的键域来排除不匹配的文件。

为了高效地把记录键和布隆过滤器进行比对，即尽量减少过滤器的读取和均衡执行器间的工作量，Hudi缓存了输入记录并使用了自定义分区器和统计规律来解决数据的偏斜。有时，如果布隆过滤器的伪正率过高，查询会增加数据的打乱操作。Hudi支持动态布隆过滤器（设置hoodie.bloom.index.filter.type=DYNAMIC_V0）。它可以根据文件里存放的记录数量来调整大小从而达到设定的伪正率。

在不久的将来，我们计划引入一个更快的布隆索引机制。该机制会在Hudi内部的元数据表中跟踪记录布隆过滤器和取值范围从而进行快速的点查询。这可以避免因从基文件中读取布隆过滤器和取值范围而导致的查询的局限性。（总体设计请参考RFC-15）

作业场景：对事件表的去重

事件流无处不在。从Apache Kafka或其他类似的消息总线发出的事件数通常是事实表大小的10-100倍。事件通常把时间（到达时间、处理时间）作为首类处理对象，比如物联网的事件流、点击流数据、广告曝光数等等。由于这些大部分都是仅追加的数据，插入和更新只存在于最新的几个分区中。由于重复事件可能发生在整个数据管道的任一节点，在存放到数据湖前去重是一个常见的需求。

上图演示了事件表的更新分布情况。

总的来说，低消耗去重一个非常挑战的工作。尽管甚至可以用一个键值存储来实现去重（即HBase索引），但索引存储的消耗会随着事件数增长而线性增长以至于变得不可行。事实上，有范围剪枝功能的布隆索引是最佳的解决方案。我们可以利用作为首类处理对象的时间来构造由事件时间戳和事件id（event_ts+event_id)组成的键，这样插入的记录就有了单调增长的键。这会在最新的几个分区里大幅提高剪枝文件的效益。

作业场景：对维度表的随机更删

这种类型的表通常包含高纬度的数据和数据链接，比如用户资料、商家信息等。这些都是高保真度的表。它们的更新量通常很小但所接触的分区和数据文件会很多，范围涉及从旧到新的整个数据集。有时因为没有很好的分区条件，这些表也会不分区。

上图演示了维度表的更新分布情况。

正如之前提到的，如果范围比较不能剪枝许多文件的话，那么布隆索引并不能带来很好的效益。在这样一个随机写入的作业场景下，更新操作通常会触及表里大多数文件从而导致布隆过滤器依据输入的更新对所有文件标明真正(true positive)。最终会导致，即使采用了范围比较，也还是检查了所有文件。使用简单索引对此场景更合适，因为它不采用提前的剪枝操作，而是直接和所有文件的所需字段连接。如果额外的运维成本可以接受的话，也可以采用HBase索引，其对这些表能提供更加优越的查询效率。

当使用全局索引时，用户也可以考虑通过设置hoodie.bloom.index.update.partition.path=true或者hoodie.simple.index.update.partition.path=true（Global Bloom）hoodie.hbase.index.update.partition.path=true（HBase）或者hoodie.hbase.index.update.partition.path=true 或者来处理分区路径需要更新的情况；例如对于以所在城市分区的用户表，会有用户迁至另一座城市的情况。这些表也非常适合采用Merge-On-Read表型。

将来我们计划在Hudi内实现记录层的索引机制，以此提高索引查询效率，同时也将省去维护额外系统（比如HBase）的开销。

总结

若没有索引功能，Hudi就不可能在超大扩展规模上实现更删操作。希望这篇文章为目前的索引机制提供了足够的背景知识和对不同权衡取舍的解释。

以下是一些颇具意义的相关工作：

• 基于Apache Flink并建立在RocksDB状态存储上的索引机制将带来真正意义上的数据湖流式插入更新。
• 全新的元数据索引将基于Hudi元数据全面翻新现有的布隆索引机制。
• 记录层的索引实现，用另一个Hudi表作为二级索引。

在接下来的开发中，项目组会对这个领域保持积极的投入。我们始终期待更多贡献者的加入以及推进路线图中的项目。如果有意参与，欢迎与我们的社区联系。

Flink SQL是Flink API的最顶层抽象，在使用Flink SQL的时候，是否被其便捷和高效惊艳到？

到底一条SQL语句是如何运行起来的？提到Flink SQL，离不开SQL引擎框架 – Calcite。Calcite是面向Hadoop的查询引擎，提供了SQL解析、优化、多重数据源查询的基础框架。Flink借助Calcite实现了SQL解析、优化和graph生成。

Flink API与Flink SQL简介

Flink提供了三层API抽象，每一层API都是便捷性和变现力之前的权衡，来应对不同的计算场景。

Flink SQL是最顶层的抽象，这层抽象在语义和程序表达式上都类似于Table API，但是程序实现都是SQL表达式:

1. SQL和Table API都是遵循关系模型：元数据和操作
   • 元数据类似于关系型数据库中的schema
   • 操作类似于关系型数据库中的操作, 如select、project、join、group-by和aggregate
2. SQL和Table API都是声明式定义，而不会执行执行的具体代码
3. 简洁，通过UDF扩展，但比core API的表达能力差
4. 执行之前，通过优化器中的优化规则对用户编写的表达式进行优化
5. TableEnvironment是Flink SQL和Table API的入口，可以无缝切换到DataStream/DataSet, 允许混用。

Calcite

Calcite简介

是一个动态数据的管理框架，可以用来构建数据库系统的语法解析模块

• 不包含数据存储、数据处理等功能
• 可以通过编写 Adaptor 来扩展功能，以支持不同的数据处理平台
• Flink SQL 使用并对其扩展以支持 SQL 语句的解析和验证

Calcite提供了SQL parser、SQL validation、Query optimizer、SQL generator和Data federator

查询的执行过程

分四步:

1. Parse(SQL -> SqlNode ): 使用JavaCC生成的parser来转换查询
2. validate(SqlNode -> SqlNode ): 通过元数据验证查询
3. optimize: 逻辑计划优化和转换为物理计划
   • 语义分析(SqlNode -> RelNode/RexNode ): 根据 SqlNode 及元信息构建 RelNode 树，也就是最初版本的逻辑计划（Logical Plan）；
   • 逻辑计划优化(RelNode -> RelNode ): 优化器的核心，根据前面生成的逻辑计划按照相应的规则（Rule）进行优化；
4. execute: 物理计划转换为应用框架的执行逻辑(如Flink的graph)

Catalog

定义Calcite查询的命名空间:

Schema : schema和table的集合，可以任意嵌套

Table: 代表单个数据集，字段定义为RelDataType

RelDataType: 代表数据集中的字段，支持所有的SQL数据类型，包括结构体

Statistic: 提供用于优化的表统计信息, 如行数、分布信息、是否为key

SQL parser

• LL(K) parser通过JavaCC(Java Compiler Compiler)写的
• 输入的查询转换为AST(abstract syntax tree)
• SqlNode代表Token
• SqlNode可以通过unparse方法转换会SQL

SqlNode代表AST中的一个节点

SqlDialect代表特定数据库的方言规则

Query optimizer

查询计划(Query plans)代表执行一个查询必须的步骤

Query优化:

• 优化逻辑计划
• 目标通常是尽量减少计划中必须在早期处理的数据量
• 将逻辑计划转换为物理计划
• 物理计划与引擎有关，代表了物理执行过程

常见的优化方法:

核心概念

1. 关系代数（Relational algebra）：即关系表达式。它们通常以动词命名，例如 Sort, Join, Project, Filter, Scan, Sample.
2. 行表达式（Row expressions）：例如 RexLiteral (常量), RexVariable (变量), RexCall (调用) 等，例如投影列表（Project）、过滤规则列表（Filter）、JOIN 条件列表和 ORDER BY 列表、WINDOW 表达式、函数调用等。使用 RexBuilder 来构建行表达式。
3. 表达式有各种特征（Trait）：使用 Trait 的 satisfies() 方法来测试某个表达式是否符合某 Trait 或 Convention.
4. 转化特征（Convention）：属于 Trait 的子类，用于转化 RelNode 到具体平台实现（可以将下文提到的 Planner 注册到 Convention 中）. 例如 JdbcConvention，FlinkConventions.DATASTREAM 等。同一个关系表达式的输入必须来自单个数据源，各表达式之间通过 Converter 生成的 Bridge 来连接。
5. 规则（Rules）：用于将一个表达式转换（Transform）为另一个表达式。它有一个由 RelOptRuleOperand 组成的列表来决定是否可将规则应用于树的某部分。

Planner(规划器)

规划器（Planner） ：即请求优化器，它可以根据一系列规则和成本模型（例如基于成本的优化模型 VolcanoPlanner、启发式优化模型 HepPlanner）来将一个表达式转为语义等价（但效率更优）的另一个表达式。

HepPlanner(启发式优化模型)

• 与Spark优化器类似的启发式优化器
• 启发式优化比CBO要快速
• 如果规则对计划做出相反的改变，则存在无限递归的风险

VolcanoPlanner(基于成本的优化模型)

• 遍历所有的规则，选择代价最小的计划
• 代价是通过关系表达式提供的
• 并不是所有可能的计划都会计算
• 当经过指定的迭代未显著提升将停止优化
• 代价包括行数、I/O和CPU
• Statistics用来提高代价评估的准确性
• Calcite提供了工具来在统计资源消耗

Flink 与 Calcite

Calcite中Flink中的重要作用

在Flink中，Calcite扮演着重要的角色：

• 以Calcite Catalog为核心，上面承载了Table/SQL API
• Flink SQL和Table的代码最后生成Calcite Logic Plan(SqlNode Tree)
• 随后验证、优化为 RelNode 树，
• 最终通过 Rules（规则）和 Convention（转化特征）生成具体的 DataSet Plan（批处理）或 DataStream Plan（流处理），即 Flink 算子构成的处理逻辑。

Table / SQL API 的编程框架如下：

• 通过 TableEnvironment 配置 CalciteConfig 对象，自动设置 SQL & Table API 默认处理参数。

• 使用 registerTableSource() 来将一个 TableSource 注册到 rootSchema. 后续可以通过 scan() 获取此 Table 并调用各种 Table API 进行处理。

• 接下可以调用 sqlQuery() 和 sqlUpdate() 方法来使用 SQL 语句进行数据处理。

Flink SQL的执行流程

Planner接口: 解析SQL，转换为Transformation

Executor接口: 将Planner转换的Transformation生成streamGraph并执行

Parser接口: 负责SQL解析，parse方法将SQL语句转换为Operation数组

• 通过Calcite将SQL解析为SqlNode
• 根据SqlNode的类型，将SqlNode转换为Operation数组

DDL语句的转换过程:

SqlNode转换为RelNode:

• 推断Table类型
• 推断计算列
• 推断watermark分配

SQL转换:

1. Operation->RelNode
2. 优化RelNode
3. RelNode->ExecNode
4. ExecNode->Transformation算子

死码

[参考文献]

1. [Parsing database Query with Apache Calcite](Parsing database Query with Apache Calcite - Knoldus Blogs)
2. Apache Calcite: One planner fits all (slideshare.net)