Flink 流批一体在字节跳动的探索与实践

三叶草

背景

字节跳动旗下拥有今日头条、抖音等多款产品，每天服务着数亿用户，由此产生的数据量和计算量也是很大的：

  
• EB 级别海量的存储空间
• 每天平均 70PB 数据的增量
• 每秒钟百万次数的实时推荐请求
• 超过 400 万核的流式计算资源、500 万核的批式计算资源
  

这对我们的整个架构，包括计算架构和存储架构都带来了巨大的挑战。
业务困境

如上图所示，左边是一个非常典型，业界应用也很多的数据链路图。这个数据链路是一个典型的 Lamda 架构，整个数据链路分为批式计算链路和流式计算链路。
在字节跳动内部，通常需要批式计算和流式计算两条链路共同服务于下游的应用。

  
• 在批式计算链路中，我们主要应用 Spark 引擎，通过 Spark 引擎在批式存储中拿到数据，经过 ETL 的计算后，存入下游的存储，从而服务下游的应用。
• 流式计算链路，也是我们整个实时推荐、实时信息流的核心链路。我们会通过消息中心件把实时数据进行缓存存入，然后运用 Flink 实时计算引擎进行处理，处理后经过消息中间件的缓存传输存入下游的存储，来服务下层的应用。
  

整个计算架构分成两条链路，带来了两个比较严重的问题：

  
• 计算不同源
  
  
     
  • 维护成本高。批式计算主要使用 Spark 引擎，流式计算使用 Flink 引擎。维护两套引擎就意味着使用两套代码，工程师的维护成本和学习成本都非常高。
  • 数据一致性和质量难以保障。两套代码之间不能相互复用，所以数据的一致性和数据的质量难以保障。
  • 无法混合调度造成资源浪费。批式计算和流式计算的高峰期是不同的。对流式计算来说，用户的使用高峰期一般是白天或凌晨12点之前，那么这些时间段也就是流式计算的高峰，此时对计算资源的需求是非常高的。相对而言，批式计算对运算时间并没有严格的限制，比如可以在凌晨12点之后到早上6、7点之间进行大量运算。所以，如果流式计算和批式计算的资源无法进行混合调度，那么就无法对运算资源进行错峰使用，造成资源的巨大浪费。
    
     
• 存储不同源
  
  
     
  • 数据不一致，维护成本高。如果两条链路同时服务于下游应用的话，那么两套存储系统也是分隔开的，依然存在数据不一致的问题。同时，维护流式、批式两套存储系统的成本也非常高。
    
    
  

针对上述困境，在字节跳动内部，我们选择了流批一体的解决方案。
什么是流批一体
那么，什么是流批一体呢？

  
• 从计算层面来讲，就是用同一个引擎、同一套代码及同样的 API ，同时处理有限的数据流和无限的数据流，同时应对在线处理和离线处理（其中有限数据的处理对应离线处理，而无限数据的处理则对应在线处理），达到降本增效的目的。
• 在存储方面，流批一体即存储系统能够同时满足流式数据和批式数据的存储，并能够有效地进行协同以及元数据信息的更新。
  

架构体系使用流批一体后，数据流向如下图左边流程图所示。

无论是流式数据还是批式数据，都可以直接或经过简单加工后存入统一存储中。而后，使用流批一体统一的计算引擎进行 ETL 计算，再服务下游的应用。由此，整个流批一体的架构实质上实现了计算同源和存储同源。

  
• 计算同源。用一套代码、一套逻辑去处理流式任务和批式任务，达到了降本增效的目的，同时也大幅提升了资源利用率。
• 存储同源。在存储方面统一存储，避免了存储资源的浪费，同时也在很大的程度上避免了数据不一致。
  

  字节跳动的流批一体实践
在字节跳动，我们使用 Flink 作为流批一体统一的计算引擎，Iceberg 作为流批一体统一的存储方式。简单的数据流向如下图。

在上游取到信息后，根据 Binlog 信息，使用 BMQ（字节跳动自研的云原生消息队列引擎） 也就是消息中间件产品，将数据实时传输到流批一体计算引擎 Flink 中，进行流式处理或批式处理后，将整个数据 更新到 Iceberg 数据湖。数据湖的存储底座也是字节跳动自研的存储底座——大数据文件存储（CloudFS）。
为什么选择 Flink
我们为什么会选择 Flink 作为流批一体的计算引擎呢？
主要原因在于，Flink 是一个面向有限流和无限流有状态计算的分布式计算框架，它能够支持流处理和批处理两种应用类型。
在传统意义上，Flink 是一个无限的数据流。但如果我们用一个个的时间窗口把无限的数据流进行切分，我们就得到很多有限数据流，对 Flink 来说，批式数据只是流式数据的一种特例。

无论是无限数据流还是有限处理流，Flink 都可以通过同一种 API、同一套代码进行处理之后，服务下游的数据。这样的流程也可以极大地减少工程师的学习和维护成本。

可以说，Flink 无论是从上层的代码层面、SDK 层面、API 层面，还是下层的调度器层面，都是针对流批一体的整体架构来进行设计的，是可以从上至下完整地支持流批一体的数据处理引擎。

Flink 流批一体架构

推荐系统流批一体实践
下面以字节跳动的推荐系统为例，向大家阐述字节跳动内部使用流批一体的典型实践。
推荐系统在字节跳动占据着重要的位置。今日头条的新闻、抖音的视频，每一条信息流都需要由推荐系统进行推荐。如前文所述，整个推荐系统每天承载着庞大的推荐任务量和数据量。
在推荐系统的整个数据处理链路中，流式处理和批式处理都占据着重要的位置。尤其是在特征计算模块，推荐系统需要为用户实时地推荐信息流，保证实时性和准确性，同时也需要进行模型训练以提升推荐准确性。双数据链路的设计带来了诸多问题。
双链路存在的核心问题

推荐系统数据链路抽象图

在流式链路中，我们接收用户请求，获得用户的实时在线特征，这些实时在线特征经过实时的流式处理之后，再结合在线特征库，就可以得到一个比较庞大的特征组。随后，将整个特征组输入到在线预测模型中，就可以得到预测的结果，从而实时地为用户推荐信息流。
同时，这些特征也会被存入离线存储（如 HDFS）中，后续会利用这些特征进行线下的批式模型训练。对于离线训练来说，存入 HDFS 中的数据，经过批式的 ETL 处理后，输入到离线的模型训练中，训练出的模型可以用于更新在线服务的模型，从而更准确地服务用户。
然而，正如上文所述，推荐系统的数据链路分了在线和离线两个体系，所以推荐系统在计算和使用在线特征和离线特征时，需要分别使用两种不同计算引擎和存储进行在、离线特征处理，带来了以下问题：

  
• 对流处理和批处理分别维护两套代码，业务成本过高
• 特征回溯难度大
• 如何使用历史数据初始化状态难定义
• 数据不统一，存储成本高
  

  Flink SQL 实现计算一体
针对这些业务困境和核心问题，我们使用了 Flink SQL 去实现整个计算的流批一体。在整个数据处理链路中，我们基于 Flink 引擎，使用 Flink SQL 的方式同时处理流式任务和批式任务，由此可以达到：

  
• 同时支持 Unbounded、Bounded 数据源
• 支持 Join 和 Union
• 流批一体的执行模式
• 自定义统一 Sink Connectors
  

通过 Flink SQL 实现流批一体后，整个数据链路在计算的速度、特征的迭代，及业务降本增效上都取得了极大的成果。主要原因在于使用 Flink SQL 实现流批一体后：

  
• 同一份代码既可以实时计算，又可以批式计算
• 节省开发成本，加速特征迭代过程
  

如上图所示，推荐系统中的特征需要定期回溯并用以更新推荐模型，保证在线推荐的准确性。使用 Flink SQL 实现了流批计算一体后，我们可以用同一套代码去进行实时计算和批式计算，批式计算可以使用与实时计算同样的代码进行历史数据的回溯，这就保证了数据一致。
Iceberg 实现存储一体
在存储方面，我们选用了 Iceberg 作为统一的存储格式。如下图所示，特征数据经过字节跳动自研的消息队列引擎 BMQ 统一地流入 Flink 引擎，在 Flink SQL 进行处理之后，再 Upsert 到整个数据库当中，进行统一的管理。

基于 Iceberg 实现特征的统一存储，具备以下能力：

  
• 存储流批一体，支持元数据的更新和管理
• 提供 ACID 保证和快照功能
• 并发读写
• 计算存储引擎解耦
• Arrow 向量化数据传输
• 小文件 Compaction
  

  优化收益
从整体业务收益来看，采用 Flink + Iceberg 的流批一体架构后，取得了较为明显的降本增效效果：

  
• 维护一套数据处理代码，人力成本大幅降低
• 特征存储成本降低 40% 以上
• Arrow 数据传输进行特征训练，CPU 消耗降低 13%，网络 IO 降低 40%
  

  云原生计算流批一体解决方案
云原生计算团队将字节跳动内部流批一体方案进行整合优化后，输出了云原生计算平台——一个开箱即用的、基于 Kubernets 的大数据 & AI 开发平台。
云原生计算平台部署灵活，既能以火山引擎的公有云为底座，也能以专有云及其他的 Kubernets 底座进行部署。
在火山引擎资源底座的基础之上，我们还提供丰富的资源调度策略、自动化流水线的 CICD 交付，以及丰富的资源管理、数据管理、作业管理等功能。

云原生计算平台架构

在此之上，是字节跳动流批一体解决方案的核心引擎。
首先是流批一体的存储。流批一体存储主要是由两部分组成，一部分是火山引擎自研的大数据统一存储 CloudFS——作为整个存储层和数据加速层为上游的引擎提供服务。另一部分是 Iceberg，我们以 Iceberg 为存储层，利用上层的 Table Format 进行元数据信息的管理。与此同时，通过对数据和源数据的操作，增加整个数据流数据的管控性和流转速度。
其次是三款计算引擎。

  
• Flink 实时计算引擎。我们在整个链路中会把 Flink 作为流批一体的引擎。
• Spark 批式计算引擎。Spark 其实也是一款流批一体的计算引擎，在批式计算有它独特的优势。
• Ray 动态引擎。Ray 动态引擎相对较新。我们用整个 Ray 动态引擎来做资源的极致扩缩、极致弹性，服务数据挖掘场景。
• 在三款主要的计算引擎之外，还有字节跳动自研的云原生消息引擎 BMQ，及开放搜索引擎 Open Search。
  

通过这五款引擎，我们打造了一个端到端的数据链路——数据存入大数据统一文件存储（CloudFS）之后，经由不同的引擎进行处理，服务上层业务。
平台管控台 UI 及大数据开发平台统一管理数据处理过程，同时整个云原生计算平台生态开放，可以对接各种大数据开发平台以及 AI 开发的 Studio IDE。
最上层是应用层。由主引擎及存储组成的流批一体解决方案，可以形成数据可视化、安全及金融风控、数据化运营等解决方案，端到端地服务数字营销，实时大屏、车联网等业务场景。
总的来说，在云原生计算平台流批一体解决方案中，我们选择了 Flink 作为流批一体的计算引擎，CloudFS 和 Iceberg 作为流批一体的统一存储，服务机器学习场景和数据处理场景，无论是字节内部的推荐系统，还是对外部提供服务，都能够针对这两种场景提供完备的服务。