2021年，字节跳动产品的总MAU已经超过19亿。在以Tik Tok、今日头条、西瓜视频为代表的产品业务背景下，一个强大的推荐系统显得尤为重要。Flink提供了非常强大的SQL模块和有状态计算模块。目前在字节推荐场景中，实时简单计数特性、窗口计数特性、序列特性已经完全迁移到Flink SQL方案中。将Flink SQL与Flink有状态计算能力相结合，我们正在构建下一代通用统一架构的基本特征计算，希望能够高效支持常见有状态和无状态基本特征的产生。

工作经历

对于字节跳动旗下的今日头条、Tik Tok、西瓜视频等产品，基于Feed流和短期效果的推荐是核心业务场景。推荐系统最基本的燃料是特性，高效生产的基本特性对商业推荐系统的迭代至关重要。

主要业务场景

抖音、火山短视频等为代表的短视频应用推荐场景，例如 Feed 流推荐、关注、社交、同城等各个场景，整体在国内大概有 6 亿+规模 DAU；头条、西瓜等为代表的 Feed 信息流推荐场景，例如 Feed 流、关注、子频道等各个场景，整体在国内数亿规模 DAU；

业务难点和挑战

目前，字节跳动推荐场景基本特征的制作现状是“百花齐放”。离线特征计算的基本模式是基于Spark和Flink计算引擎，通过消费Kafka、BMQ、Hive、HDFS、Abase、RPC等数据源实现特征计算，然后将特征结果写入在线和离线存储。不同类型的基础特征计算分散在不同的服务中，缺乏业务抽象，导致运维成本大，稳定性问题多。

更重要的是，缺乏统一的基础特征生产平台，使得业务特征开发的迭代速度和维护不方便。比如业务端需要自己维护大量离线任务，特色生产环节缺乏监控，无法满足不断发展的业务需求。

在字节业务的规模下，构建统一的实时特征生产系统面临着巨大的挑战，主要来自四个方面:

庞大的业务规模:Tik Tok、今日头条、西瓜、火山等产品的数据规模可达日均PB级。例如，在Tik Tok的场景中，晚高峰Feed的播放量达到数百万QPS，客户端上报的用户行为数据达到数千万IOPS。业务期望在任何时候，功能任务都能保持流动，消费无滞后等。，这就要求特征制作具有非常高的稳定性。

实时特征要求更高:在以直播、电商、短视频为代表的推荐场景中，为了保证推荐效果，实时特征的离线制作时效需要稳定在分钟。

更好的可扩展性和灵活性:随着业务场景的日益复杂，功能需求更加灵活多变。从统计、系列、属性类型的特征生产，到窗口特征、多维特征的灵活支持，业务端需要特征中心支持新的特征类型和逐渐衍生的需求。

服务迭代速度快:特征中心站提供的面向服务的DSL需要足够多的场景，特征生产环节让服务编写尽可能少的代码，底层计算引擎和存储引擎对服务完全透明，完全释放服务计算、存储选择和优化的负担，完全实现实时基础特征的规模化生产，不断提升特征生产力；

迭代进化过程

在字节业务的爆发式增长中，为了满足各种业务特性的需求，很多特性服务都是从推荐场景中衍生出来的。这些服务在特定的业务场景和历史条件下，很好地支持了业务的快速发展。一般流程如下:

推荐场景特征服务的演进

2020年初，是一个重要的节点。我们开始将Flink SQL和Flink State技术系统引入到特征生产中，并逐步在计数特征系统、模型训练的样本拼接、窗口特征等场景中实现。，探索新一代专题节目制作方案的思路。

新一代系统架构

结合上述业务背景，我们重新设计了基于Flink SQL和Flink状态计算能力的新一代实时特征计算方案。新方案的定位是:解决基础特征计算和在线服务，提供更抽象的基础特征服务层DSL。在计算层，基于Flink SQL灵活的数据处理和表达能力，以及Flink State状态存储和计算能力，支持各种复杂的窗口计算。大大缩短业务基础功能的制作周期，提高功能输出环节的稳定性。在新的体系结构中，我们将特征产生的环节分为数据源提取/拼接、状态存储和计算三个阶段。Flink SQL完成特征数据提取和流式拼接，Flink State完成特征计算的中间状态存储。

有状态特征是非常重要的一类特征，其中最常用的特征是带有各种窗口的特征，比如统计最近5分钟播放VV的视频。对于以字节为单位的窗口类型的特征，有一些基于存储引擎的方案。总体思路是“轻离线，重在线”，即所有的窗口状态存储和特征聚合计算都放在存储层，在线完成。离线数据流负责基础数据的过滤和写入，离线详细数据按照时间划分进行汇总和存储(类似于微批量)。底层存储多为KV存储或专门优化的存储引擎，线上层完成复杂的窗口聚合计算逻辑。每个请求到来后，线上层拉取存储层的详细数据进行聚合计算。

我们新的解决方案是“轻在线，重离线”，即我们把所有重的时间片细节数据状态存储和窗口聚合计算都放在离线层。结果聚合由离线窗口触发机制完成，特征结果推送到在线KV存储。在线模块非常轻量，只负责简单的在线服务，大大简化了在线层的架构复杂度。在离线存储层。我们主要依靠Flink提供的native state存储引擎RocksDB，充分利用离线计算集群的本地SSD磁盘资源，大大减轻了在线KV存储的资源压力。

对于长窗口的特性(7天以上的窗口特性)，由于涉及到Flink状态层详细数据的回溯过程，Flink嵌入式状态存储引擎没有提供特别好的外部数据回注机制(或者说不适合做)。所以对于这种“状态冷启动”的场景，我们引入集中式存储作为底层状态存储层的存储介质，整体结构为混合架构。比如7天内的状态存储在本地SSD，7 ~ 30天内的状态存储在集中存储引擎。离线数据回溯可以非常方便地写入集中存储。

除了窗口特征之外，这种机制也适用于其他类型的有状态特征(例如序列特征)。

实时特征分类系统

特征类型

定义

功能示例

状态特征

有状态特性是一种非常重要的特性。我们将有状态特性定义为:计算特性需要缓存上下文数据。

带有窗口的特征，例如抖音视频最近1h的点赞量（滑动窗口）、直播间用户最近一个 session 的看播时长（session 窗口）等；序列特征，例如最近100个推荐展现视频。

无状态特征

简单的ETL特性，可以通过简单的数据过滤来计算的特性。

模型预测特征

需要由外部复杂模型预测的特征

用户的年龄、性别和其他特征。

图形特征

直播和社交关系场景中需要两跳关系的图类型有很多特点。

许多图形特征也是有状态特征。

礼物排序：用户观看最多的主播收到最多的礼物，首选需要找到用户观看最多的主播 ArchorId，然后通过 archon_id 获取到主播收到最多的礼物 id；社交关系：好友（可能是挖掘出来的关系）关注、看播、送礼、连麦的房间，社交关系天然是图数据结构。

整体架构

数据源层

在新的集成特性架构中，我们将各种数据源抽象为Schema表，因为底层的Flink SQL计算引擎层为数据源提供了非常友好的表格式抽象。在推荐场景中，依赖的数据源非常多样，每个特性的上游都依赖于一个或多个数据源。数据源可以是卡夫卡、RMQ、KV存储和RPC服务。对于多数据源，支持数据源的流式和批量拼接。拼接类型包括基于键粒度的窗口连接和窗口联合连接。维度表连接支持base、RPC、HIVE等。每种类型的具体拼接逻辑如下:

数据源类型

图式分析

卡夫卡、BMQ

卡夫卡、BMQ等消息类型基本都是JSON和PB，都是自描述的数据类型。可以非常方便地映射到SchemaTable格式，其中对于PB类型，业务需要上传PB IDL来完成表模式定义。

KV存储

KV存储中的值大多是JSON和PB格式，类似于MQ。业务通过提供PB IDL完成表模式的定义。通过FlinkSQL的维度表连接能力，我们将获取外部存储数据源的通用流程抽象为基本的维度表连接操作，简化了业务开发周期。

位置遥控(remote position control)

FlinkSQL提供了对rpc维度表的连接能力，业务提供了RPC Thrift IDL完整的RPC响应表模式定义。通过维度表连接，我们将通过RPC获取外部数据源的通用流程抽象为基本的维度表连接模型，简化了业务开发周期。

储备

Hive本身就是SchemaTable的存储格式。对于有少量在线Join数据的离线Hive数据(其实就是MapSide Join)，可以通过Hive维度表Join来实现。

可以组合三种类型的Join和Union来实现复杂的多流拼接。例如(联合b)窗口连接(c查找连接d)。

拼接类型

拼接逻辑

评论

窗口连接

使用Flink native API提供的Join操作符来连接落入同一个窗口的多个数据流的数据。

使用TumblingWindow直接对原始数据流进行分段，按照event_time或process_time对齐两个窗口，然后关联数据。

基于关键粒度的区间状态连接

类似于样本拼接逻辑。通过Union上游的多个数据源，在每个关联的主键上注册timer，等待固定的时间窗口完成多个数据源的Join操作。

区间状态连接是对状态存储的数据的再处理。两个上游数据流经过Union后，同一个uid的实例数据和标签数据落在同一个运算符中，Joiner中的正负样本就是通过这种连接方式生成的。

查找维度表联接

通过关联主键，从诸如base、RPC、Hive等服务中查看要关联的数据。，并完成数据连接操作。

多数据源联合

联合多个数据源

此外，Flink SQL支持复杂字段的计算能力，即业务端可以根据数据源定义的TableSchema基本字段计算扩展字段。商业计算逻辑的本质是UDF。我们会把UDF API接口提供给业务端，然后把JAR上传到功能后台进行加载。此外，对于简单的计算逻辑，后台还通过提交简单的Python代码来支持多语言计算。

服务DSL

从业务角度提供高度抽象的特性生产DSL语言，屏蔽底层计算和存储引擎的细节，让业务端专注于业务特性的定义。DSL层提供:数据源、数据格式、数据提取逻辑、数据生成特征类型、数据输出方式等。

状态存储层

如上所述，新的特性集成方案解决的主要痛点是:如何处理各种类型(一般是滑动窗口)的具有状态特性的计算问题。对于这类特征，离线计算层架构中会有一个状态存储层，提取层提取的RawFeature会按照片槽存储(片可以是时间片，也可以是会话片等。).切片类型在内部是一种接口类型，在架构上可以根据业务需求自行扩展。实际上，状态并不存储原始的RawFeature(它浪费了存储原始行为数据的存储时间空)，而是一个转换成FeaturePayload的POJO结构，它支持各种常见的数据结构类型:

Int：存储简单的计数值类型(多维度 counter)；HashMap：存储二维计数值，例如 Action Counter，key 为 target_id，value 为计数值；SortedMap: 存储 topk 二维计数 ；LinkedList： 存储 id_list 类型数据；HashMap>：存储二维 id_list；自定义类型，业务可以根据需求 FeaturePayload 里面自定义数据类型

状态更新的业务接口:输入是SQL提取/拼接层提取的RawFeature，业务端可以根据业务需求通过updateFeatureInfo接口更新状态层。对于常用的特性类型，内置了更新接口，业务端的自定义特性类型可以继承更新接口的实现。

/** * 特征状态update接口 */public interface FeatureStateApi extends Serializable { /** * 特征更新接口, 上游每条日志会提取必要字段转换为fields, 用来更新对应的特征状态 * * @param fields * context: 保存特征名称、主键 和 一些配置参数; * oldFeature: 特征之前的状态 * fields: 平台/配置文件 中的抽取字段 * @return */FeaturePayLoad assign(Context context,FeaturePayLoad feature, Map<String, Object> rawFeature);}

复制代码

当然，无状态ETL特性不需要状态存储层。

计算层

特征层完成特征计算的聚合逻辑，状态特征计算的输入数据是状态存储层存储切片的FeaturePayload对象。的简单ETL特性没有状态存储层，输入直接是SQL提取层的data RawFeature对象。具体界面如下:

/** * 有状态特征计算接口 */public interface FeatureStateApi extends Serializable { /** * 特征聚合接口，会根据配置的特征计算窗口, 读取窗口内所有特征状态，排序后传入该接口 * * @param featureInfos, 包含2个field * timeslot: 特征状态对应的时间槽 * Feature: 该时间槽的特征状态 * @return */ FeaturePayLoad aggregate(Context context, List<Tuple2<Slot, FeaturePayLoad>> slotStates);}

复制代码

状态特征聚合接口

/** * 无状态特征计算接口 */public interface FeatureConvertApi extends Serializable { /** * 转换接口, 上游每条日志会提取必要字段转换为fields, 无状态计算时，转换为内部的feature类型; * * @param fields * fields: 平台/配置文件 中的抽取字段 * @return */ FeaturePayLoad convert(Context context, FeaturePayLoad featureSnapshot, Map<String, Object> rawFeatures);}

复制代码

无状态特征计算接口

此外，特征计算层的执行由触发机制触发。目前支持的触发机制主要包括:

策略

解释

OnTimerTrigger

周期性触发特征的计算逻辑

OnUpdateTrigger

上游状态层的每次更新都会触发特征计算。

自定义触发器

自定义特征计算的触发时间

商务着陆

目前，在字节推荐场景中，新一代功能架构已经在Tik Tok直播、电商、推送、Tik Tok推荐场景中推出了部分实时功能。主要有有状态特征，如带窗口的一维统计型、二维倒拉链型、二维TOPK型、实时CTR/CVR率型特征、序列型特征等。

在实现核心业务指标方面取得了显著成绩。在直播场景下，基于新功能架构强大的表达能力，多项功能上线后，业务的看播核心指标和互动指标的收益非常显著。在电子商务场景中，基于新的功能架构推出400+实时功能。其中，直播电商方面，业务核心GMV和订单率指标收益显著。在Tik Tok推送场景中，基于新功能构建线下存储能力，将用户行为数据聚合后写入下游渠道进行存储，大大缓解了商家下游数据库的压力。在某些情况下，QPS可以减少到以前的10%左右。此外，Tik Tok推荐Feed、点评等服务正在基于新的功能架构重构原有的功能体系。

值得一提的是，在Tik Tok的电商和直播场景中，Flink流任务的最大状态已经达到了60T，而且这个量级还在不断增加。预计在不久的将来，单个任务的状态可能会超过100T，这对架构的稳定性是一个很大的挑战。

性能优化

Flink状态缓存

目前Flink提供了两种StateBackend:基于Heap的FileSystemStateBackend和基于RocksDB的RocksDBStateBackend。对于FileSystemStateBackend，因为数据都在内存中，所以访问速度非常快，没有额外的开销。但是，RocksDBStateBackend有磁盘查找、序列化/反序列化等额外开销，CPU使用率会明显增加。有大量的作业使用字节内的状态。对于大型状态作业，RocksDBStateBackend通常用于管理本地状态数据。RocksDB是一个KV数据库，它以LSM的形式组织数据。在实际使用过程中，它具有以下特点:

应用层和 RocksDB 的数据交互是以 Bytes 数组的形式进行，应用层每次访问都需要序列化/反序列化；数据以追加的形式不断写入 RocksDB 中，RocksDB 后台会不断进行 compaction 来删除无效数据。

业务端使用状态的场景主要是get-update。在使用RocksDB作为本地状态存储的过程中，出现了以下问题:

爬虫数据导致热 key，状态会不断进行更新(get-update)，单 KV 数据达到 5MB，而 RocksDB 追加更新的特点导致后台在不断进行 flush 和 compaction，单 task 出现慢节点（抖音直播场景）。电商场景作业多数为大状态作业(目前已上线作业状态约 60TB)，业务逻辑中会频繁进行 State 操作。在融合 Flink State 过程中发现 CPU 的开销和原有的基于内存或 abase 的实现有 40%~80%的升高。经优化后，CPU 开销主要集中在序列化/反序列化的过程中。

为了解决上述问题，我们可以通过在内存中维护一个对象缓存来优化热数据访问并减少CPU开销。通过以上背景介绍，我们希望为StateBackend提供一个通用的缓存功能，通过Flink StateBackend缓存功能的设计方案，达到以下目的:

减少 CPU 开销：通过对热点数据进行缓存，减少和底层 StateBackend 的交互次数，达到减少序列化/反序列化开销的目的。提升 State 吞吐能力：通过增加 Cache 后，State 吞吐能力应比原有的 StateBackend 提供的吞吐能力更高。理论上在 Cache 足够大的情况下，吞吐能力应和基于 Heap 的 StateBackend 近似。Cache 功能通用化：不同的 StateBackend 可以直接适配该 Cache 功能。目前我们主要支持 RocksDB，未来希望可以直接提供给别的 StateBackend 使用，例如 RemoteStateBackend。

通过与字节基础设施Flink团队的合作，在实时特性制作升级中，在线缓存的大部分场景的CPU利用率将可能高达50%左右；

PB IDL剪辑

在以字节为单位的实时特征的离线生成中，我们主要依靠Kafka作为数据流。这些卡夫卡都是PB定义的数据，有很多字段。一般公司级的大题目有100+个字段，但大部分特色制作任务只使用其中的一部分字段。对于Protobuf格式的数据源，我们可以完全剪切数据流，屏蔽掉一些不必要的字段，以节省反序列化的成本。PB类型的日志可以直接剪辑到idl，保持必要字段的序列号不变，反序列化时会跳过未知字段的解析，对CPU来说更经济，但是网络带宽不会盈利。预计裁剪可以节省大量CPU资源。PB IDL裁剪上线后，大部分任务的CPU收益在30%左右。

遇到的问题

新架构特征生产任务的本质是一个有状态的Flink任务，底层状态存储StateBackend主要是本地RocksDB。有两个难题，一个是任务DAG变化检查点失效，另一个是本地存储不能很好的支持特征状态的历史数据回溯。

实时特征任务不能动态添加新的特征：对于一个线上的 Flink 实时特征生产任务，我们不能随意添加新的特征。这是由于引入新的特征会导致 Flink 任务计算的 DAG 发生改变，从而导致 Flink 任务的 Checkpoint 无法恢复，这对实时有状态特征生产任务来说是不能接受的。目前我们的解法是禁止更改线上部署的特征任务配置，但这也就导致了线上生成的特征是不能随便下线的。对于这个问题暂时没有找到更好的解决办法，后期仍需不断探索。特征状态冷启动问题：目前主要的状态存储引擎是 RocksDB，不能很好地支持状态数据的回溯。

后续规划

目前新一代架构在字节推荐场景下快速演进，解决了实时窗口特性的制作问题。

为了实现统一推荐场景下的特征生产，我们将基于Flink SQL流程和批量集成能力，在批量特征生产方面继续发力。此外，基于胡迪数据湖技术，将完成特征实时入湖，高效支撑模型训练场景的离线特征回溯痛点。引擎方向，计划继续探索CEP，推动更多电商场景的落地实践。在实时窗口计算方向，我们将继续深入研究Flink原生窗口机制，以解决当前方案面临的窗口特征数据的退出问题。

支持批式特征：这套特征生产方案主要是解决实时有状态特征的问题，而目前字节离线场景下还有大量批式特征是通过 Spark SQL 任务生产的。后续我们也会基于 Flink SQL 流批一体的计算能力，提供对批式场景特征的统一支持，目前也初步有了几个场景的落地；特征离线入湖：基于 Hudi On Flink 支持实时特征的离线数仓建设，主要是为了支持模型训练样本拼接场景离线特征回溯；Flink CEP 规则引擎支持：Flink SQL 本质上就是一种规则引擎，目前在线上我们把 Flink SQL 作为业务 DSL 过滤语义底层的执行引擎。但 Flink SQL 擅长表达的 ETL 类型的过滤规则，不能表达带有时序类型的规则语义。在直播、电商场景的时序规则需要尝试 Flink CEP 更加复杂的规则引擎。Flink Native Windowing 机制引入：对于窗口类型的有状态特征，我们目前采用上文所述的抽象 SlotState 时间切片方案统一进行支持。另外 Flink 本身提供了非常完善的窗口机制，通过 Window Assigner、Window Trigger 等组件可以非常灵活地支持各种窗口语义。因此后续我们也会在窗口特征计算场景引入 Flink 原生的 Windowing 机制，更加灵活地支持窗口特征迭代。Flink HybridState Backend 架构：目前在字节的线上场景中，Flink 底层的 StateBackend 默认都是使用 RocksDB 存储引擎。这种内嵌的存储引擎不能通过外部机制去提供状态数据的回灌和多任务共享，因此我们需要支持 KV 中心化存储方案，实现灵活的特征状态回溯。静态属性类型特征统一管理：通过特征平台提供统一的 DSL 语义，统一管理其他外部静态类型的特征服务。例如一些其他业务团队维度的用户分类、标签服务等。

作者介绍:

郭文飞，字节跳动推荐系统基础服务方向负责人。Byte是2015年初加入的，主要负责推荐系统的基础服务方向，比如去重、计数、特性等。

字节跳动推荐架构团队的实时计算方向，负责Tik Tok、今日头条、西瓜视频等超过10亿用户的产品推荐系统架构的实时计算系统的设计与开发，保证系统的稳定性和高可用性。抽象出通用的实时计算系统，构建统一的推荐特征中心，实现灵活可扩展的高性能存储系统和计算模型，实现推荐服务的去重、计数、特征服务等先进的实时推荐数据流系统。目前缺人。欢迎追求技术的同学们加入我们，打造世界一流的先进实时推荐数据流系统。联系方式:guowenfei@bytedance.com。