[Java 探索之路～大数据篇] 新时代大数据流处理入门指南

本文主要介绍大数据基础，以及 flink 流计算

【基础知识】

1. 批处理与流处理

1.批处理

批处理（Batch Processing）是指对一批数据进行处理。我们身边的批处理比比皆是，最常见的批处理例子有：微信运动每天晚上有一个批处理任务，把用户好友一天所走的步数统计一遍，生成排序结果后推送给用户；银行信用卡中心每月账单日有一个批处理任务，把一个月的消费总额统计一次，生成用户月度账单；国家统计局每季度对经济数据做一次统计，公布季度国内生产总值（GDP）。
可见，批处理任务一般是对一段时间的数据聚合后进行处理的。对于数据量庞大的应用，如微信运动、银行信用卡中心等情景，一段时间内积累的数据总量非常大，计算非常耗时。

2.流处理

数据其实是以流（Stream）的方式持续不断地产生着的，流处理（Stream Processing）就是对数据流进行处理。
:::info
时间就是金钱，对数据流进行分析和处理，获取实时数据价值越发重要
:::
如“双十一电商大促销”，管理者要以秒级的响应时间查看实时销售业绩、库存信息以及与竞品的对比结果，以争取更多的决策时间；股票交易要以毫秒级的速度来对新信息做出响应；风险控制要对每一份欺诈交易迅速做出处理，以减少不必要的损失；网络运营商要以极快速度发现网络和数据中心的故障；等等。以上这些场景，一旦出现故障，造成服务的延迟，损失都难以估量。因此，响应速度越快，越能减少损失、增加收入。

因此，响应速度越快，越能减少损失、增加收入。而IoT和5G的兴起将为数据生成提供更完美的底层技术基础，海量的数据在IoT设备上采集，并通过高速的5G通道传输到服务器，庞大的实时数据流将汹涌而至，流处理的需求肯定会爆炸式增长。

2. 为什么需要一个优秀的流处理框架

处理实时流的系统通常被称为流计算框架、实时计算框架或流处理框架。下面就来解释为何需要一个可靠的流处理框架。

1. 股票交易的业务场景

我们都知道股票交易非常依赖各类信息，一些有可能影响股票市场价格的信息经常首发于财经网站、微博、微信等社交媒体平台上。作为人类的我们不可能24小时一直监控各类媒体，如果有一个自动化的系统来做一些分析和预警，将为决策者争取到更多时间。
假设我们有数只股票的交易数据流，我们可以通过这个数据流来计算以10秒为一个时间窗口的股票价格波动，选出那些超过5%变化幅度的股票，并将这些股票与媒体的实时文本数据做相关分析，以判断媒体上的哪些实时信息会影响股票价格。当相关分析的结果足够有说服力时，可以将这个系统部署到生产环境，实时处理股票与媒体数据，产生分析报表，并发送给交易人员。那么，如何构建一个可靠的程序来解决上述业务场景问题呢？

2.生产者——消费者模型

处理流数据一般使用“生产者-消费者”（Producer-Consumer）模型来解决问题。如图1-6所示，生产者生成数据，将数据发送到一个缓存区域（Buffer），消费者从缓存区域中消费数据。这里我们暂且不关心生产者如何生产数据，以及数据如何缓存，我们只关心如何实现消费者。

在股票交易的场景中，我们可以启动一个进程来实现消费者，该进程以10秒为一个时间窗口，统计时间窗口内的交易情况，找到波动最大的那些股票。同时，该进程也对新流入的媒体文本进行分析。这个逻辑看起来很容易实现，但深挖之后会发现问题繁多。

3. 流处理框架要解决的诸多问题

（1）可扩展性

股票交易和媒体文本的数据量都非常大，仅以微博为例，平均每秒有上千条、每天有上亿条微博数据。一般情况下，单个节点无法处理这样规模的数据，这时候需要使用分布式计算。假如我们使用类似MPI的框架，需要手动设计分治算法，这对很多程序员来说有一定的挑战性。随着数据不断增多，我们能否保证我们的程序能够快速扩展到更多的节点上，以应对更多的计算需求？具体而言，当计算需求增多时，计算资源能否线性增加而不是耗费大量的资源，程序的代码逻辑能否保持简单而不会变得极其复杂？一个具有可扩展性的系统必须能够优雅地解决这些问题。

（2）数据倾斜

在分布式计算中，数据需要按照某种规则分布到各个节点上。假如数据路由规则设计得不够完善，当数据本身分布不均匀时，会发生数据倾斜，这很可能导致部分节点数据量远大于其他节点。这样的后果是：轻则负载重的节点延迟过高，重则引发整个系统的崩溃。假如一条突发新闻在网络媒体平台引发激烈的讨论和分析，数据突增，程序很可能会崩溃。数据倾斜是分布式计算中经常面临的一个问题。

（3）容错性

整个系统崩溃重启后，之前的那些计算如何恢复？或者部分节点发生故障，如何将该节点上的计算迁移到其他的节点上？我们需要一个机制来做故障恢复，以增强系统的容错性。

（4）时序错乱

限于网络条件和其他各种潜在影响因素，流处理引擎处理某个事件的时间并不是事件本来发生的时间。比如，你想统计上午11:00:00到11:00:10的交易情况，然而发生在11:00:05的某项交易因网络延迟没能抵达，这时候要直接放弃这项交易吗？绝大多数情况下我们会让程序等待，比如我们会假设数据最晚不会延迟超过10分钟，因此程序会等待10分钟。等待一次也还能接受，但是如果有多个节点在并行处理呢？每个节点等待一段时间，最后做数据聚合时就要等待更长时间。
:::info
批处理框架一般处理一个较长时间段内的数据，数据的时序性对其影响较小。批处理框架用更长的时间来换取更好的准确性。流处理框架对时序错乱更为敏感，框架的复杂程度也因此大大增加。
:::

4. Flink是解决上述问题的最佳选择之一。

如果用Flink去解决前文提到的股票建模问题，只需要设置时间窗口，并在这个时间窗口下做一些数据处理的操作，还可以根据数据量来设置由多少节点并行处理。

3. 分布式计算

分布式计算已经有很多比较成熟的方案，其中比较有名的有消息传递接口（Message Passing Interface，MPI）和映射归约模型（MapReduce）。

MPI

MPI是一个“老牌”分布式计算框架，从MPI这个名字也可以看出，MPI主要解决节点间数据通信的问题。在前MapReduce时代，MPI是分布式计算的业界标准。MPI现在依然广泛运用于全球各大超级计算中心、大学、研究机构中，许多物理、生物、化学、能源等基础学科的大规模分布式计算都依赖MPI。图1-3所示为使用MPI在4台服务器上并行计算的示意图。

MapReduce

基于MapReduce编程模型，不同的团队分别实现了自己的大数据框架：Hadoop是较早的一种开源实现，如今已经成为大数据领域的业界标杆，之后又出现了Spark和Flink。这些框架提供了编程接口，辅助程序员存储、处理和分析大数据。

4. 大数据代表性技术

1. Hadoop

• Hadoop 生态圈的核心组件主要有如下3个。

• Hadoop MapReduce：Hadoop版本的MapReduce编程模型，可以处理海量数据，主要面向批处理。

• HDFS：HDFS（Hadoop Distributed File System）是Hadoop提供的分布式文件系统，有很好的扩展性和容错性，为海量数据提供存储支持。

• YARN：YARN（Yet Another Resource Negotiator）是Hadoop生态圈中的资源调度器，可以管理一个Hadoop集群，并为各种类型的大数据任务分配计算资源。
  :::info
  这三大组件中，数据存储在HDFS上，由MapReduce负责计算，YARN负责集群的资源管理。
  :::
  除了三大核心组件，Hadoop生态圈还有很多其他著名的组件，部分如下。

• Hive：借助Hive，用户可以编写结构化查询语言（Structured Query Language，SQL）语句来查询HDFS上的结构化数据，SQL语句会被转化成MapReduce运行。

• HBase：HDFS可以存储海量数据，但访问和查询速度比较慢，HBase可以提供给用户毫秒级的实时查询服务，它是一个基于HDFS的分布式数据库。HBase最初受Google Bigtable技术的启发。

• Kafka：Kafka是一款流处理框架，主要用作消息队列。

• ZooKeeper：Hadoop生态圈中很多组件使用动物来命名，形成了一个大型“动物园”，ZooKeeper是这个动物园的管理者，主要负责分布式环境的协调。

Hadoop生态圈

2. Spark

Spark是一款大数据处理框架，其开发初衷是改良Hadoop MapReduce的编程模型和提高运行速度，尤其是提升大数据在机器学习方向上的性能。与Hadoop相比，Spark的改进主要有如下两点。

• 易用性：MapReduce 模型比 MPI 更友好，但仍然不够方便。因为并不是所有计算任务都可以被简单拆分成 Map 和 Reduce，有可能为了解决一个问题，要设计多个 MapReduce 任务，任务之间相互依赖，整个程序非常复杂，导致代码的可读性和可维护性差。Spark提供更加方便易用的接口，提供Java、Scala、Python和R语言API，支持SQL、机器学习和图计算，覆盖了绝大多数计算场景。
• 速度快：Hadoop 的 Map 和 Reduce 的中间结果都需要存储到磁盘上，而 Spark 尽量将大部分计算放在内存中。加上 Spark 有向无环图的优化，在官方的基准测试中，Spark比Hadoop快一百倍以上。
  :::info
  Spark的核心在于计算，主要目的在于优化Hadoop MapReduce计算部分，在计算层面提供更细致的服务。
  :::
  Spark并不能完全取代Hadoop，实际上，从图可以看出，Spark融入了Hadoop生态圈，成为其中的重要一员。一个Spark任务很可能依赖HDFS上的数据，向YARN申请计算资源，将结果输出到HBase上。当然，Spark也可以不用依赖这些组件，独立地完成计算。

Spark生态圈

Spark主要面向批处理需求，因其优异的性能和易用的接口，Spark已经是批处理界绝对的“王者”。Spark的子模块 Spark Streaming 提供了流处理的功能，它的流处理主要基于 mini-batch 的思想。如图所示，Spark Streaming将输入数据流切分成多个批次，每个批次使用批处理的方式进行计算。
:::info
因此，Spark是一款集批处理和流处理于一体的处理框架。
:::

3. Kafka

Kafka也是一种面向大数据领域的消息队列框架。在大数据生态圈中，Hadoop的HDFS或Amazon S3提供数据存储服务，Hadoop MapReduce、Spark和Flink负责计算，Kafka常常用来连接不同的应用系统。
如图所示，企业中不同的应用系统作为数据生产者会产生大量数据流，这些数据流还需要进入不同的数据消费者，Kafka起到数据集成和系统解耦的作用。系统解耦是让某个应用系统专注于一个目标，以降低整个系统的维护难度。在实践上，一个企业经常拆分出很多不同的应用系统，系统之间需要建立数据流管道（Stream Pipeline）。

假如没有Kafka的消息队列，M个生产者和N个消费者之间要建立M×N个点对点的数据流管道，Kafka就像一个中介，让数据管道的个数变为M+N，大大减小了数据流管道的复杂程度。
从批处理和流处理的角度来讲，数据流经Kafka后会持续不断地写入HDFS，积累一段时间后可提供给后续的批处理任务，同时数据流也可以直接流入Flink，被用于流处理。
:::info
随着流处理的兴起，Kafka不甘心只做一个数据流管道，开始向轻量级流处理方向努力，但相比Spark和Flink这样的计算框架，Kafka的主要功能侧重在消息队列上。
:::

4. Flink

Flink是由德国3所大学发起的学术项目，后来不断发展壮大，并于2014年年末成为Apache顶级项目之一。在德语中，“flink”表示快速、敏捷，以此来表征这款计算框架的特点。Flink主要面向流处理，如果说Spark是批处理界的“王者”，那么Flink就是流处理领域冉冉升起的“新星”。流处理并不是一项全新的技术，在Flink之前，不乏流处理引擎，比较著名的有Storm、Spark Streaming，如图展示了流处理框架经历的三代演进。

• 2011年成熟的 Apache Strom（以下简称Storm）是第一代被广泛采用的流处理引擎。它是以数据流中的事件为最小单位来进行计算的。以事件为单位的框架的优势是延迟非常低，可以提供毫秒级的延迟。流处理结果依赖事件到达的时序准确性，Storm并不能保障处理结果的一致性和准确性。Storm只支持至少一次（At-Least-Once）和至多一次（At-Most-Once），即数据流里的事件投递只能保证至少一次或至多一次，不能保证只有一次（Exactly-Once）。在多项基准测试中，Storm的数据吞吐量和延迟都远逊于Flink。对于很多对数据准确性要求较高的应用，Storm有一定劣势。此外，Storm不支持SQL，不支持中间状态（State）。

• 2013年成熟的Spark Streaming是第二代被广泛采用的流处理框架。
  :::info
  Spark是“一统江湖”的大数据处理框架，Spark Streaming 采用微批次（mini-batch）的思想，将数据流切分成一个个小批次，一个小批次里包含多个事件，以接近实时处理的效果。
  :::
  这种做法保证了“Exactly-Once”的事件投递效果，因为假如某次计算出现故障，重新进行该次计算即可。Spark Streaming的API相比第一代流处理框架更加方便易用，与Spark批处理集成度较高，因此Spark可以给用户提供一个流处理与批处理一体的体验。但因为Spark Streaming以批次为单位，每次计算一小批数据，比起以事件为单位的框架来说，延迟从毫秒级变为秒级。

• 与前两代引擎不同，在2015年前后逐渐成熟的Flink是一个支持在有界和无界数据流上做有状态计算的大数据处理框架。它以事件为单位，支持SQL、状态、水位线（Watermark）等特性，支持“Exactly-Once”。比起Storm，它的吞吐量更高，延迟更低，准确性能得到保障；比起Spark Streaming，它以事件为单位，达到真正意义上的实时计算，且所需计算资源相对更少。

具体而言，Flink的优点如下。

• 支持事件时间（Event Time）和处理时间（Processing Time）多种时间语义。即使事件乱序到达，Event Time也能提供准确和一致的计算结果。Procerssing Time适用于对延迟敏感的应用。Exactly-Once投递保障。
• 毫秒级延迟。
• 可以扩展到上千台节点、在阿里巴巴等大公司的生产环境中进行过验证。易用且多样的API，包括核心的 DataStream API 和 DataSet API 以及Table API和SQL。可以连接大数据生态圈各类组件，包括Kafka、Elasticsearch、JDBC、HDFS和Amazon S3。
• 可以运行在Kubernetes、YARN、Mesos和独立（Standalone）集群上。

5. 大数据处理平台的演进

前文已经提到，流处理框架经历了3代的更新迭代，大数据处理也随之经历了从Lambda架构到Kappa架构的演进。

Lambda架构

当以Storm为代表的第一代流处理框架成熟后，一些互联网公司为了兼顾数据的实时性和准确性，采用图所示的Lambda架构来处理数据并提供在线服务。Lambda架构主要分为3部分：批处理层、流处理层和在线服务层。其中数据流来自Kafka这样的消息队列。

1. 批处理层

在批处理层，数据流首先会被持久化保存到批处理数据仓库中，积累一段时间后，再使用批处理引擎来进行计算。这个积累时间可以是一小时、一天，也可以是一个月。处理结果最后导入一个可供在线应用系统查询的数据库上。批处理层中的批处理数据仓库可以是HDFS、Amazon S3或其他数据仓库，批处理引擎可以是MapReduce或Spark。
批处理层能保证某份数据的结果的准确性，而且即使程序运行失败，直接重启即可。此外，批处理引擎一般扩展性好，即使数据量增多，也可以通过增加节点数量来横向扩展。

2. 流处理层

很明显，假如整个系统只有一个批处理层，会导致用户必须等待很久才能获取计算结果，一般有几小时的延迟。电商数据分析部门只能查看前一天的统计分析结果，无法获取当前的结果，这对实时决策来说是一个巨大的时间鸿沟，很可能导致管理者错过最佳决策时机。因此，在批处理层的基础上，Lambda架构增加了一个流处理层，用户行为日志会实时流入流处理层，流处理引擎生成预处理结果，并导入一个数据库。分析人员可以查看前一小时或前几分钟内的数据结果，这大大增强了整个系统的实时性。
:::info
但数据流会有事件乱序等问题，使用早期的流处理引擎，只能得到一个近似准确的计算结果，相当于牺牲了一定的准确性来换取实时性。
:::
早期的流处理引擎有一些缺点，由于准确性、扩展性和容错性的不足，流处理层无法直接取代批处理层，只能给用户提供一个近似结果，还不能为用户提供一个一致准确的结果。因此Lambda架构中，出现了批处理和流处理并存的现象。

3. 在线服务层

在线服务层直接面向用户的特定请求，需要将来自批处理层准确但有延迟的预处理结果和流处理层实时但不够准确的预处理结果做融合。
:::info
在融合过程中，需要不断将流处理层的实时数据覆盖批处理层的旧数据。
:::
很多数据分析工具在数据合并上下了不少功夫，如Apache Druid，它可以融合流处理与批处理结果。当然，我们也可以在应用程序中人为控制预处理结果的融合。存储预处理结果的数据库可能是关系型数据库MySQL，也可能是Key-Value键值数据库Redis或HBase。

4. Lambda架构的优缺点

Lambda架构在实时性和准确性之间做了一个平衡，能够解决很多大数据处理的问题。
它的优点如下。批处理的准确度较高，而且在数据探索阶段可以对某份数据试用不同的方法，反复对数据进行实验。另外，批处理的容错性和扩展性较强。流处理的实时性较强，可以提供一个近似准确的结果。
Lambda架构的缺点也比较明显。使用两套大数据处理引擎，如果两套大数据处理引擎的API不同，有任何逻辑上的改动，就需要在两边同步更新，维护成本高，后期迭代的时间周期长。早期流处理层的结果只是近似准确。

Kappa架构

Kafka的创始人杰·克雷普斯认为在很多场景下，维护一套Lambda架构的大数据处理平台耗时耗力，于是提出在某些场景下，没有必要维护一个批处理层，直接使用一个流处理层即可满足需求，即图所示的Kappa架构。

Kappa架构的兴起主要有如下两个原因。

• Kafka可以保存更长时间的历史数据，它不仅起到消息队列的作用，也可以存储数据，替代数据库。
• Flink流处理引擎解决了事件乱序下计算结果的准确性问题。

Kappa架构相对更简单，实时性更好，所需的计算资源远小于Lambda架构，随着实时处理需求的不断增长，更多的企业开始使用Kappa架构。
Kappa架构的流行并不意味着不再需要批处理，批处理在一些特定场景上仍然有自己的优势。比如，进行一些数据探索、机器学习实验，需要使用批处理来反复验证不同的算法。
:::info
Kappa架构适用于一些逻辑固定的数据预处理流程，比如统计一个时间段内商品的曝光和购买次数、某些关键词的搜索次数等，这类数据处理需求已经固定，无须反复试验迭代。
:::
Flink以流处理见长，但也实现了批处理的API，是一个集流处理与批处理于一体的大数据处理引擎，为Kappa架构提供更可靠的数据处理性能，未来Kappa架构将在更多场景下逐渐替换Lambda架构。

6. 流处理的基本概念

在某些场景下，流处理打破了批处理的一些局限。Flink作为一款以流处理见长的大数据引擎，相比其他流处理引擎具有众多优势。本节将对流处理的一些基本概念进行细化，这些概念是入门流处理的必备基础，至此你将正式进入数据流的世界。

1. 延迟和吞吐

• 延迟
  :::info
  延迟表示一个事件被系统处理的总时间，一般以毫秒为单位。根据业务不同，我们一般关心平均延迟（Average Latency）和分位延迟（Percentile Latency）。
  :::

假设一个食堂的自助取餐流水线是一个流处理系统，每个就餐者前来就餐是它需要处理的事件，从就餐者到达食堂到他拿到所需菜品并付费离开的总耗时，就是这个就餐者的延迟。如果正赶上午餐高峰期，就餐者极有可能排队，这个排队时间也要算在延迟中。

例如，99分位延迟表示对所有就餐者的延迟进行统计和排名，取排名第99%位的就餐者延迟。

一般商业系统更关注分位延迟，因为分位延迟比平均延迟更能反映这个系统的一些潜在问题。
还是以食堂的自助餐流水线为例，该流水线的平均延迟可能不高，但是在就餐高峰期，延迟一般会比较高。如果延迟过高，部分就餐者会因为等待时间过长而放弃排队，用户体验较差。通过检查各模块分位延迟，能够快速定位到哪个模块正在“拖累”整个系统的性能。
:::info
延迟对于很多流处理系统非常重要，比如欺诈检测系统、告警监控系统等。Flink可以将延迟降到毫秒级别。如果用mini-batch的思想处理同样的数据流，很可能有分钟级到小时级的延迟，因为批处理引擎必须等待一批数据达到才开始进行计算。
:::

• 吞吐
  :::info
  吞吐表示一个系统最多能处理多少事件，一般以单位时间处理的事件数量为标准。
  :::
  需要注意的是，吞吐除了与引擎自身设计有关，也与数据源发送过来的事件数据量有关，有可能流处理引擎的最大吞吐量远小于数据源的数据量。
  比如，自助取餐流水线可能在午餐时间的需求最高，很可能出现大量排队的情况，但另外的时间几乎不需要排队等待。假设一天能为1 000个人提供就餐服务，共计10小时，那么它的平均吞吐量为100人/小时；仅午间2小时的高峰期就提供了600人，它的峰值吞吐量是300人/小时。比起平均吞吐量，峰值吞吐量更影响用户体验，如果峰值吞吐量低，会导致就餐者等待时间过长而放弃排队。排队的过程被称作缓存（Buffering）。如果排队期间仍然有大量数据进入缓存，很可能超出系统的极限，就会出现反压（Backpressure）问题，这时候就需要一些优雅的策略来处理类似问题，否则会造成系统崩溃，用户体验较差。

• 延迟与吞吐

延迟与吞吐其实并不是相互孤立的，它们相互影响。如果延迟高，那么很可能吞吐较低，系统处理不了太多数据。
为了优化这两个指标，首先提高自助取餐流水线的行进速度，加快取餐各个环节的进程。当用户量大到超过流水线的瓶颈时，需要再增加一个自助取餐流水线。这就是当前大数据系统都在采用的两种加速方式，第一是优化单节点内的计算速度，第二是使用并行策略，分而治之地处理数据。如果一台计算机做不了或做得不够快，那就用更多的计算机一起来做。
:::info
综上，延迟和吞吐是衡量流处理引擎的重要指标。如何保证流处理系统保持高吞吐和低延迟是一项非常有挑战性的工作。
:::

2. 窗口与时间

1. 不同窗口模式

比起批处理，流处理对窗口（Window）和时间概念更为敏感。
在批处理场景下，数据已经按照某个时间维度被分批次地存储了。一些公司经常将用户行为日志按天存储，一些开放数据集都会说明数据采集的时间始末。因此，对于批处理任务，处理一个数据集，其实就是对该数据集对应的时间窗口内的数据进行处理。
**在流处理场景下，数据以源源不断的流的形式存在，数据一直在产生，没有始末。**我们要对数据进行处理时，往往需要明确一个时间窗口，比如，数据在“每秒”“每小时”“每天”的维度下的一些特性。窗口将数据流切分成多个数据块，很多数据分析都是在窗口上进行操作，比如连接、聚合以及其他时间相关的操作。
:::info
如图展示了3种常见的窗口形式：滚动窗口、滑动窗口、会话窗口。

:::

• 滚动窗口（Tumbling Window）模式一般定义一个固定的窗口长度，长度是一个时间间隔，比如小时级的窗口或分钟级的窗口。窗口像车轮一样，滚动向前，任意两个窗口之间不会包含同样的数据。
• 滑动窗口（Sliding Window）模式也设有一个固定的窗口长度。假如我们想每分钟开启一个窗口，统计10分钟内的股票价格波动，就使用滑动窗口模式。当窗口的长度大于滑动的间隔，可能会导致两个窗口之间包含同样的事件。其实，滚动窗口模式是滑动窗口模式的一个特例，滚动窗口模式中滑动的间隔正好等于窗口的大小。
• 会话窗口（Session Window）模式的窗口长度不固定，而是通过一个间隔来确定窗口，这个间隔被称为会话间隔（Session Gap）。当两个事件之间的间隔大于会话间隔，则两个事件被划分到不同的窗口中；当事件之间的间隔小于会话间隔，则两个事件被划分到同一窗口。

会话（Session）本身是一个用户交互概念，常常出现在互联网应用上，一般指用户在某App或某网站上短期内产生的一系列行为。比如，用户在手机淘宝上短时间大量的搜索和点击的行为，这系列行为事件组成了一个会话。接着可能因为一些其他因素，用户暂停了与App的交互，过一会用户又使用App，经过一系列搜索、点击、与客服沟通，最终下单。

2. 时间语义

• Event Time 和 Processing Time

“时间”是平时生活中最常用的概念之一，在流处理中需要额外注意它，因为时间的语义不仅与窗口有关，也与事件乱序、触发计算等各类流处理问题有关。常见的时间语义如下。
:::info

• Event Time：事件实际发生的时间。

• Processing Time：事件被流处理引擎处理的时间。
  :::
  对于一个事件，自其发生起，Event Time 就已经确定不会改变。因各类延迟、流处理引擎各个模块先后处理顺序等因素，不同节点、系统内不同模块、同一数据不同次处理都会产生不同的 Processing Time。

• “一分钟”真的是一分钟吗？

在很多应用场景中，时间有着不同的语义，“一分钟”真的是一分钟吗？很多手机游戏中多玩家在线实时竞技，假设我们在玩某款手机游戏，该游戏将数据实时发送给游戏服务器，服务器计算一分钟内玩家的一些操作，这些计算影响用户该局游戏的最终得分。
当游戏正酣，我们进入了电梯，手机信号丢失，一分钟后才恢复信号；幸好手机在电梯期间缓存了掉线时的数据，并在信号恢复后将缓存数据传回了服务器，如图展示了这个场景的流处理过程。在丢失信号的这段时间，你的数据没有被计算进去，显然这样的计算不公平。当信号恢复时，数据重传到服务器，再根据 Event Time 重新计算一次，那就非常公平了。我们可以根据 Event Time 复现一个事件序列的实际顺序。因此，使用Event Time是最准确的。

• Watermark

虽然使用Event Time更准确，但问题在于，因为各种不可控因素，事件上报会有延迟，那么最多要等待多长时间呢？从服务器的角度来看，在事件到达之前，我们也无法确定是否有事件发生了延迟，如何设置等待时间是一个很难的问题。比如刚才的例子，我们要统计一分钟内的实时数据，考虑到事件的延迟，如何设置合理的等待时间，以等待一分钟内所有事件都到达服务器？

也正因为这个问题，流处理与批处理在准确性上有差距，因为批处理一般以更长的一段时间为一个批次，一个批次内延迟上报的数据比一个流处理时间窗口内延迟上报的数据相对更少。比如电商平台上，对于计算一件商品每分钟点击次数，使用一天的总数除以分钟数，比使用一分钟时间窗口实时的点击次数更准确。可以看到，数据的实时性和准确性二者不可兼得，必须取一个平衡。

:::info
Watermark 是一种折中解决方案，它假设某个时间点上，不会有比这个时间点更晚的上报数据。
:::
当流处理引擎接收到一个 Watermark 后，它会假定之后不会再接收到这个时间窗口的内容，然后会触发对当前时间窗口的计算。比如，一种Watermark策略等待延迟上报的时间非常短，这样能保证低延迟，但是会导致错误率上升。在实际应用中，Watermark设计为多长非常有挑战性。还是以手机游戏为例，系统不知道玩家这次掉线的原因是什么，可能是在穿越隧道，可能是有事退出了该游戏，还有可能是坐飞机进入飞行模式。
那既然Event Time似乎可以解决一切问题，为什么还要使用Processing Time？前文也提到了，为了处理延迟上报或事件乱序，需要使用一些机制来等待，这样会导致延迟提高。某些场景可能对准确性要求不高，但是对实时性要求更高，在这些场景下使用Processing Time就更合适一些。

3. 状态与检查点

:::info
状态是流处理区别于批处理的特有概念。
:::

• 如果我们对一个文本数据流进行处理，把英文大写字母都改成英文小写字母，这种处理是无状态的，即系统不需要记录额外的信息。
• 如果我们想统计这个数据流一分钟内的单词出现次数，一方面要处理每一瞬间新流入的数据，另一方面要保存之前一分钟内已经进入系统的数据，额外保存的数据就是状态。

如图展示了无状态和有状态两种不同类型的计算。

状态在流处理中经常被用到。再举一个温度报警的例子，当系统在监听到“高温”事件后10分钟内又监听到“冒烟”事件，系统必须及时报警。在这个场景下，流处理引擎把“高温”事件作为状态记录下来，并判断这个状态接下来十分钟内是否有“冒烟”事件。
流处理引擎在数据流上做有状态计算主要有以下挑战。
设计能够管理状态的并行算法极具挑战。前文已经多次提到，大数据需要在多节点上分布式计算，一般将数据按照某个Key进行切分，将相同的Key切分到相同的节点上，系统按照Key维护对应的状态。如果状态数据不断增长，最后就会造成数据爆炸。因此可使用一些机制来限制状态的数据总量，或者将状态数据从内存输出到磁盘或文件系统上，持久化保存起来。系统可能因各种错误而出现故障，重启后，必须能够保证之前保存的状态数据也能恢复，否则重启后很多计算结果有可能是错误的。
:::info
检查点（Checkpoint)机制其实并不是一个新鲜事物，它广泛存在于各类计算任务上，主要作用是将中间数据保存下来。
:::
当计算任务出现问题，重启后可以根据Checkpoint中保存的数据重新恢复任务。在流处理中，Checkpoint主要保存状态数据。

4. 数据一致性保障

流处理任务可能因为各种原因出现故障，比如数据量暴涨导致内存溢出、输入数据发生变化而无法解析、网络故障、集群维护等。事件进入流处理引擎，如果遇到故障并重启，该事件是否被成功处理了呢？一般有如下3种结果。

• At-Most-Once：每个事件最多被处理一次，也就是说，有可能某些事件直接被丢弃，不进行任何处理。这种投递保障最不安全，因为一个流处理系统完全可以把接收到的所有事件都丢弃。
• At-Least-Once：无论遇到何种状况，流处理引擎能够保证接收到的事件至少被处理一次，有些事件可能被处理多次。例如，我们统计文本数据流中的单词出现次数，事件被处理多次会导致统计结果并不准确。
• Exactly-Once：无论是否有故障重启，每个事件只被处理一次。Exactly-Once意味着事件不能有任何丢失，也不能被多次处理。比起前两种保障，Exactly-Once的实现难度非常高。如遇故障重启，Exactly-Once就必须确认哪些事件已经被处理、哪些还未被处理。
  :::info
  Flink在某些情况下能提供Exactly-Once的保障。
  :::

5. 编程语言

• Flink 的核心代码由Java和Scala编写，为这两种语言提供丰富强大的API，程序员可根据自己和团队的习惯从Java和Scala中选择。
• python
• sql

【Flink】

• flink官网 https://flink.apache.org/zh/what-is-flink/flink-architecture/
• flink 中文版 https://nightlies.apache.org/flink/flink-docs-release-1.18/zh/docs/try-flink/local_installation/
• 历史版本https://nightlies.apache.org/flink/
• 官方视频 https://www.youtube.com/watch?v=0cJ565r2FVI
• 书籍 【flink 原理与实践】 https://weread.qq.com/web/reader/51032ac07236f8e05107816

是什么

Apache Flink 是一个框架和分布式处理引擎，用于在无边界和有边界数据流上进行有状态的计算。Flink 能在所有常见集群环境中运行，并能以内存速度和任意规模进行计算。

接下来，我们来介绍一下 Flink 架构中的重要方面。

处理无界和有界数据

任何类型的数据都可以形成一种事件流。信用卡交易、传感器测量、机器日志、网站或移动应用程序上的用户交互记录，所有这些数据都形成一种流。
数据可以被作为 无界 或者 有界 流来处理。

无界流

有定义流的开始，但没有定义流的结束。它们会无休止地产生数据。无界流的数据必须持续处理，即数据被摄取后需要立刻处理。我们不能等到所有数据都到达再处理，因为输入是无限的，在任何时候输入都不会完成。处理无界数据通常要求以特定顺序摄取事件，例如事件发生的顺序，以便能够推断结果的完整性。

有界流

有定义流的开始，也有定义流的结束。有界流可以在摄取所有数据后再进行计算。有界流所有数据可以被排序，所以并不需要有序摄取。有界流处理通常被称为批处理

Apache Flink 擅长处理无界和有界数据集 精确的时间控制和状态化使得 Flink 的运行时(runtime)能够运行任何处理无界流的应用。有界流则由一些专为固定大小数据集特殊设计的算法和数据结构进行内部处理，产生了出色的性能。

快速使用

1. 打开 https://flink.apache.org/downloads/ 下载 flink

因为书籍介绍的是 1.12版本的，为避免不必要的问题，下载相同版本

2. 解压

 tar -xzvf flink-1.11.2-bin-scala_2.11.tgz

3. 启动 flink

./bin/start-cluster.sh

4. 打开 flink web 页面 localhost:8081

5. 编写结合 Kafka 词频统计程序

具体参考 https://weread.qq.com/web/reader/51032ac07236f8e05107816k1f032c402131f0e3dad99f3?

package org.example;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.util.Properties;

public class WordCountKafkaInStdOut {

    public static void main(String[] args) throws Exception {

        // 设置Flink执行环境 
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        // Kafka参数 
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "flink-group");
        String inputTopic = "Shakespeare";
        String outputTopic = "WordCount";

        // Source 
        FlinkKafkaConsumer<String> consumer =
            new FlinkKafkaConsumer<String>(inputTopic, new SimpleStringSchema(),
                properties);
        DataStream<String> stream = env.addSource(consumer);

        // Transformation 
        // 使用Flink  API对输入流的文本进行操作 
        // 按空格切词、计数、分区、设置时间窗口、聚合 
        DataStream<Tuple2<String, Integer>> wordCount = stream
            .flatMap((String line, Collector<Tuple2<String, Integer>> collector) -> {
                String[] tokens = line.split("\\s");
                // 输出结果  
                for (String token : tokens) {
                    if (token.length() > 0) {
                        collector.collect(new Tuple2<>(token, 1));
                    }
                }
            })
            .returns(Types.TUPLE(Types.STRING, Types.INT))
            .keyBy(0)
            .timeWindow(Time.seconds(5))
            .sum(1);

        // Sink 
        wordCount.print();

        // execute 
        env.execute("kafka streaming word count");

    }
} 

6. 打包应用(当然在这之前需要本地调试一下，至少得运行通吧😄)
7. 使用Flink提供的命令行工具flink，将打包好的作业提交到集群上。命令行的参数 --class 用来指定哪个主类作为入口。

./bin/flink run --class org.example.WordCountKafkaInStdOut xxtarget/flink_study-1.0-SNAPSHOT.jar

class 建议直接拷贝引用

8. web 页面查看作业提交成功

9. kafka 生产者随便发点消息

10. 查看作业日志,词频统计结果

应用场景🎬

Apache Flink 功能强大，支持开发和运行多种不同种类的应用程序。它的主要特性包括：批流一体化、精密的状态管理、事件时间支持以及精确一次的状态一致性保障等。
Flink 不仅可以运行在包括 YARN、 Mesos、Kubernetes 在内的多种资源管理框架上，还支持在裸机集群上独立部署。在启用高可用选项的情况下，它不存在单点失效问题。
事实证明，Flink 已经可以扩展到数千核心，其状态可以达到 TB 级别，且仍能保持高吞吐、低延迟的特性。世界各地有很多要求严苛的流处理应用都运行在 Flink 之上。

应用场景

1. 事件驱动型应用

什么是事件驱动型应用？

事件驱动型应用是一类具有状态的应用，它从一个或多个事件流提取数据，并根据到来的事件触发计算、状态更新或其他外部动作。
事件驱动型应用是在计算存储分离的传统应用基础上进化而来。在传统架构中，应用需要读写远程事务型数据库。
相反，事件驱动型应用是基于状态化流处理来完成。在该设计中，数据和计算不会分离，应用只需访问本地（内存或磁盘）即可获取数据。系统容错性的实现依赖于定期向远程持久化存储写入 checkpoint。下图描述了传统应用和事件驱动型应用架构的区别。

事件驱动型应用的优势？

事件驱动型应用无须查询远程数据库，本地数据访问使得它具有更高的吞吐和更低的延迟。而由于定期向远程持久化存储的 checkpoint 工作可以异步、增量式完成，因此对于正常事件处理的影响甚微。
事件驱动型应用的优势不仅限于本地数据访问。传统分层架构下，通常多个应用会共享同一个数据库，因而任何对数据库自身的更改（例如：由应用更新或服务扩容导致数据布局发生改变）都需要谨慎协调。反观事件驱动型应用，由于只需考虑自身数据，因此在更改数据表示或服务扩容时所需的协调工作将大大减少。

Flink 如何支持事件驱动型应用？

事件驱动型应用会受制于底层流处理系统对时间和状态的把控能力，Flink 诸多优秀特质都是围绕这些方面来设计的。它提供了一系列丰富的状态操作原语，允许以精确一次的一致性语义合并海量规模（TB 级别）的状态数据。
此外，Flink 还支持事件时间和自由度极高的定制化窗口逻辑，而且它内置的 ProcessFunction 支持细粒度时间控制，方便实现一些高级业务逻辑。同时，Flink 还拥有一个复杂事件处理（CEP）类库，可以用来检测数据流中的模式。
Flink 中针对事件驱动应用的明星特性当属 savepoint。
:::info
Savepoint 是一个一致性的状态映像，它可以用来初始化任意状态兼容的应用。在完成一次 savepoint 后，即可放心对应用升级或扩容，还可以启动多个版本的应用来完成 A/B 测试。
:::

2. 数据分析应用

什么是数据分析应用？

数据分析任务需要从原始数据中提取有价值的信息和指标。传统的分析方式通常是利用批查询，或将事件记录下来并基于此有限数据集构建应用来完成。为了得到最新数据的分析结果，必须先将它们加入分析数据集并重新执行查询或运行应用，随后将结果写入存储系统或生成报告。
借助一些先进的流处理引擎，还可以实时地进行数据分析。和传统模式下读取有限数据集不同，流式查询或应用会接入实时事件流，并随着事件消费持续产生和更新结果。这些结果数据可能会写入外部数据库系统或以内部状态的形式维护。仪表展示应用可以相应地从外部数据库读取数据或直接查询应用的内部状态。
如下图所示，Apache Flink 同时支持流式及批量分析应用。

流式分析应用的优势？

和批量分析相比，由于流式分析省掉了周期性的数据导入和查询过程，因此从事件中获取指标的延迟更低。
不仅如此，批量查询必须处理那些由定期导入和输入有界性导致的人工数据边界，而流式查询则无须考虑该问题。
另一方面，流式分析会简化应用抽象。批量查询的流水线通常由多个独立部件组成，需要周期性地调度提取数据和执行查询。如此复杂的流水线操作起来并不容易，一旦某个组件出错将会影响流水线的后续步骤。而流式分析应用整体运行在 Flink 之类的高端流处理系统之上，涵盖了从数据接入到连续结果计算的所有步骤，因此可以依赖底层引擎提供的故障恢复机制。

Flink 如何支持数据分析类应用？

Flink 为持续流式分析和批量分析都提供了良好的支持。具体而言，它内置了一个符合 ANSI 标准的 SQL 接口，将批、流查询的语义统一起来。无论是在记录事件的静态数据集上还是实时事件流上，相同 SQL 查询都会得到一致的结果。同时 Flink 还支持丰富的用户自定义函数，允许在 SQL 中执行定制化代码。如果还需进一步定制逻辑，可以利用 Flink DataStream API 和 DataSet API 进行更低层次的控制。

3. 数据管道应用

什么是数据管道？

提取-转换-加载（ETL）是一种在存储系统之间进行数据转换和迁移的常用方法。ETL 作业通常会周期性地触发，将数据从事务型数据库拷贝到分析型数据库或数据仓库。
数据管道和 ETL 作业的用途相似，都可以转换、丰富数据，并将其从某个存储系统移动到另一个。但数据管道是以持续流模式运行，而非周期性触发。因此它支持从一个不断生成数据的源头读取记录，并将它们以低延迟移动到终点。例如：数据管道可以用来监控文件系统目录中的新文件，并将其数据写入事件日志；另一个应用可能会将事件流物化到数据库或增量构建和优化查询索引。
下图描述了周期性 ETL 作业和持续数据管道的差异。

数据管道的优势？

和周期性 ETL 作业相比，持续数据管道可以明显降低将数据移动到目的端的延迟。此外，由于它能够持续消费和发送数据，因此用途更广，支持用例更多。

Flink 如何支持数据管道应用？

很多常见的数据转换和增强操作可以利用 Flink 的 SQL 接口（或 Table API）及用户自定义函数解决。如果数据管道有更高级的需求，可以选择更通用的 DataStream API 来实现。Flink 为多种数据存储系统（如：Kafka、Kinesis、Elasticsearch、JDBC数据库系统等）内置了连接器。同时它还提供了文件系统的连续型数据源及数据汇，可用来监控目录变化和以时间分区的方式写入文件。

【Kafka】

• Kafka 官网 https://kafka.apache.org/documentation/

是什么

快速使用

:::info
注意新开命令行接口
:::

1. 打开 https://kafka.apache.org/downloads下载最新版本
   

2. 解压

tar -xzvf kafka_2.12-3.7.0.tgz

3. 启动 zookeeper 服务

./bin/zookeeper-server-start.sh config/zookeeper.properties

4. 启动 Kafka server

./bin/kafka-server-start.sh config/server.properties

5. 创建 Kafka 主题 Shakespeare

 bin/kafka-topics.sh --create --bootstrap-server localhost:9092  --replication-factor 1 --partitions 1 --topic Shakespeare

6. 查看 Kafka 主题列表

bin/kafka-topics.sh --list --bootstrap-server localhost:9092 Shakespeare

7. 启动生产者

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Shakespeare

8. 启动命令行消费者

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic Shakespeare --from-beginning

文章来源地址https://www.toymoban.com/news/detail-838213.html

到了这里，关于[Java 探索之路～大数据篇] 新时代大数据流处理入门指南的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！