摘要：本文由趣头条数据平台负责人王金海分享，主要介绍趣头条 Flink-to-Hive 小时级场景和 Flink-to-ClickHouse 秒级场景，内容分为以下四部分：
趣头条查询的页面分为离线查询页面和实时查询页面。趣头条今年所实现的改造是在实时查询中接入了 ClickHouse 计算引擎。根据不同的业务场景，实时数据报表中会展现数据指标曲线图和详细的数据指标表。目前数据指标的采集和计算为每五分钟一个时间窗口，当然也存在三分钟或一分钟的特殊情况。数据指标数据全部从 Kafka 实时数据中导出，并导入 ClickHouse 进行计算。
这时需要有一个程序监控当前 Flink 任务的数据时间已经消费到什么时候，如9点的数据，落地时需要查看 Kafka 中消费的数据是否已经到达9点，然后在 Hive 中触发分区写入。
趣头条的实时集群和离线集群是独立的，离线集群有多套，实时集群目前有一套。通过实时集群写入离线集群，会产生 HDFS nameservices 问题。在实时集群中将所有离线集群的 nameservices 用 namenode HA 的方式全部打入实时集群并不合适。那么如何在任务中通过实时集群提交到各个离线集群？
实时要写入离线 HDFS，可能会涉及用户权限问题。实时提交的用户已经定义好该用户在所有程序中都是同一个用户，但离线中是多用户的，因此会造成实时和离线用户不对等。趣头条在 API 中添加了 withBucketUser 写 HDFS。配置好 nameservices后，接下来只需要知道该 HDFS 路径通过哪个用户来写，比如配置一个 stream 用户写入。
API 层级的好处是一个 Flink 程序可以指定多个不同的 HDFS 和不同的用户。多用户写入的实现是在 Hadoop file system 中加一个 ugi.do as ，代理用户。以上为趣头条使用 Flink 方式进行实时数据同步到 Hive 的一些工作。其中可能会出现小文件问题，小文件是后台程序进行定期 merge，如果 CheckPoint 间隔时间较短，如3分钟一次，会出现大量小文件问题。
趣头条目前有很多实时指标，平均每五分钟或三分钟计算一次，如果每一个实时指标用一个 Flink 任务，或者一个 Flink SQL 来写，比如消费一个 Kafka Topic，需要计算其日活、新增、流程等等当用户提出一个新需求时，需要改当前的 Flink 任务或者启动一个新的 Flink 任务消费 Topic。
因此会出现 Flink 任务不断修改或者不断起新的 Flink 任务的问题。趣头条尝试在 Flink 后接入 ClickHouse，实现整体的 OLAP。下图为秒级实现架构图。从 Kafka 到 Flink，到 Hive，到 ClickHouse 集群，对接外部 Horizon（实时报表），QE（实时 adhoc 查询），千寻（数据分析），用户画像（实时圈人）。
ClickHouse 采用列式存储 +LZ4、ZSTD 数据压缩。其次，计算存储结合本地化+向量化执行。Presto 数据可能存储在 Hadoop 集群或者 HDFS 中，实时拉取数据进行计算。而 ClickHouse 计算存储本地化是指每一台计算机器存在本地 SSD 盘，只需要计算自己的数据，再进行节点合并。同时，LSM merge tree+Index。将数据写入 ClickHouse 之后，会在后台开始一个线程将数据进行 merge，做 Index 索引。如建常见的 DT 索引和小时级数据索引，以提高查询性能。第四，SIMD+LLVM 优化。SIMD 是单指令多数据集。第五，SQL 语法及 UDF 完善。ClickHouse 对此有很大需求。在数据分析或者维度下拽时需要更高的特性，如时间窗口的一部分功能点。
Flink 导入 ClickHouse，在数据查询或展示报表时，会遇到一些问题，比如 Flink 任务出现故障、报错或数据反压等，或 ClickHouse 集群出现不可响应，zk 跟不上，insert 过快或集群负载等问题，这会导致整个任务出现问题。
如果流数据量突然暴增，启动 Flink 可能出现一段时间内不断追数据的情况，需要进行调整并行度等操作帮助 Flink 追数据。但这时已经出现数据积压，若还要加大 Flink 并发度处理数据，ClickHouse 限制 insert 不能过快，否则会导致恶性循环。因此当 Flink 故障或 ClickHouse 集群故障时，等待 ClickHouse 集群恢复后，Flink 任务从最新数据开始消费，不再追过去一段时间的数据，通过 Hive 将数据导入到 ClickHouse。
由于之前已经通过 Kafka 将数据实时落地到 Hive，通过 Hive 将数据写入 ClickHouse 中。ClickHouse 有分区，只需要将上一个小时的数据删除，导入 Hive 的一小时数据，就可以继续进行数据查询操作。Backfill 提供了 Flink 任务小时级容错以及 ClickHouse 集群小时级容错机制。
Flink 是流批一体计算引擎，但是没有流批一体的存储。趣头条会用 HBase、Kudu、Redis 等能够与 Flink 实时交互的 KV 存储进行数据计算。如计算新增问题，目前趣头条的方案是需要将 Hive 历史用户刷到 Redis 或 HBase 中，与 Flink 进行实时交互判断用户是否新增。
但因为 Hive 中的数据和 Redis 中的数据是存储为两份数据。其次 Binlog 抽取数据会涉及 delete 动作，Hbase，Kudu 支持数据修改，定期回到 Hive 中。带来的问题是 HBase，Kudu 中存在数据，Hive 又保存了一份数据，多出一份或多份数据。如果有流批一体的存储支持上述场景，当 Flink 任务过来，可以与离线数据进行实时交互，包括实时查询 Hive 数据等，可以实时判断用户是否新增，对数据进行实时修改、更新或 delete，也能支持 Hive 的批的动作存储。
未来，趣头条考虑对 Flink 做流批的存储，使 Flink 生态统一为流批结合。
王金海，10 年互联网历练，先后在唯品会负责用户画像系统，提供人群的个性化营销服务；饿了么担任架构师，负责大数据任务调度、元数据开发、任务画像等工作；现为趣头条数据中心平台负责人，负责大数据基础计算层（spark、presto、flink、clickhouse）、平台服务层（libra 实时计算、kepler 离线调度）、数据产品层（qe即时查询、horizon 数据报表、metadata 元数据、数据权限等）、以及团队建设。
版权声明：本文中所有内容均属于阿里云开发者社区所有，任何媒体、网站或个人未经阿里云开发者社区协议授权不得转载、链接、转贴或以其他方式复制发布/发表。申请授权请邮件已获得阿里云开发者社区协议授权的媒体、网站，在转载使用时必须注明稿件来源：阿里云开发者社区，原文作者姓名，违者本社区将依法追究责任。 如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件至：un.com 进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容。
一套基于Apache Flink构建的一站式、高性能实时大数据处理平台，广泛适用于流式数据处理、离线数据处理、DataLake计算等场景。