网站源文件修改,微信seo什么意思,做dna胎儿亲子鉴定网站,如何处理公关危机简介#xff1a; 将 Flink 无缝地集成到以 Airflow 和 Hive 为主的批处理系统的技术挑战和应对方案。 本文介绍了 SmartNews 利用 Flink 加速 Hive 日表的生产#xff0c;将 Flink 无缝地集成到以 Airflow 和 Hive 为主的批处理系统的实践。详细介绍过程中遇到的技术挑战和应…简介 将 Flink 无缝地集成到以 Airflow 和 Hive 为主的批处理系统的技术挑战和应对方案。 本文介绍了 SmartNews 利用 Flink 加速 Hive 日表的生产将 Flink 无缝地集成到以 Airflow 和 Hive 为主的批处理系统的实践。详细介绍过程中遇到的技术挑战和应对方案以供社区分享。主要内容为 项目背景问题的定义项目的目标技术选型技术挑战整体方案及挑战应对项目成果和展望后记一、项目背景
SmartNews 是一家机器学习驱动的互联网公司。自 2012 年于日本东京成立并在美国和中国设有办公室。经过 8 年多的发展SmartNews 已经成长为日本排名第一美国成长最快的新闻类应用覆盖全球超过 150 多个国家市场。据 2019 年初统计SmartNews 的 iOS 和 Android 版本全球累计下载量已经超过 5000 万次。
SmartNews 在过去 9 年的时间基于 Airflow, Hive, EMR 等技术栈构建了大量的数据集。随着数据量的增长这些离线表的处理时间在逐渐拉长。另外随着业务方迭代节奏的加快对表的实时性也提出了更高的要求。因此SmartNews 内部发起了 Speedy Batch 的项目以加快现有离线表生产效率。
本次分享便是 Speedy Batch 项目中的一个例子加速用户行为 (actions) 表的实践。
APP 端上报的用户行为日志每日通过 Hive 作业生成日表这个表是许多其他表的源头至关重要。这个作业需要运行 3 个小时进而拉高了许多下游表的延迟 (Latency)明显影响数据科学家、产品经理等用户的使用体验。因此我们需要对这些作业进行提速让各个表能更早可用。
公司业务基本上都在公有云上服务器的原始日志以文件形式上传至云存储按日分区目前的作业用 Airflow 调度到 EMR 上运行生成 Hive 日表数据存储在云存储。
二、问题的定义
1. 输入
新闻服务器每隔 30 秒上传一个原始日志文件文件上传至相应日期和小时的云存储目录。
2. 输出
原始日志经过 ETL 处理之后按日 (dt) 和行为 (action) 两级分区输出。action 种类约 300 个不固定常有增减。
3. 用户
对这个表的使用是广泛的多途径的。有从 Hive 里查询也有从 PrestoJupyter 和 Spark 里查询我们甚至不能确定以上就是全部的访问途径。
三、项目的目标
将 actions 表的时延从 3 小时缩短至 30 分钟 对下游用户保持透明。透明又分两个方面 功能方面用户无需修改任何代码做到完全无感性能方面新项目产生的表不应该导致下游读取时的性能下降
四、技术选型
在本项目之前同事已经对该作业做了多轮次改进效果不是很显著。
尝试过的方案包括增加资源投入更多的机器但遇到了云存储的 IOPS 限制每个 prefix 最多支持 3000 个并发读写这个问题在输出阶段尤为明显即多个 reducer 同时向同一个 action 子目录输出的时候容易碰到这个限制。另外还尝试了按小时预处理然后到每日凌晨再合并成日表但合并过程亦耗时较多整体时延还是在 2.5 小时左右效果不够显著。
鉴于服务器端的日志是近实时上传至云存储团队提出了流式处理的思路摒弃了批作业等待一天、处理 3 小时的模式而是把计算分散在一整天进而降低当天结束后的处理用时。团队对 Flink 有比较好的背景加上 Flink 近期对 Hive 的改进较多因此决定采用基于 Flink 的方案。
五、技术挑战
挑战是多方面的。
1. 输出 RC 文件格式
当前 Hive 表的文件格式为 RCFile为了保证对用户的透明我们只能在现有的 Hive 表上做 in-place 的 upgrade也就是我们得重用当前表那么 Flink 输出的文件格式也得符合 RCFile 格式因为一张 Hive 表只能有一个格式。
RCFile 属于 bulk format (相对应的是 row format)在每次 checkpoint 时必须一次性输出。如果我们选择 5 分钟一次 checkpoint那么每个 action 每 5 分钟必须输出一个文件这会大量增加结果文件数进而影响下游的读取性能。特别是对于低频 action文件数会上百倍的增加。我们了解了 Flink 的文件合并功能但那是在一个 checkpoint 内多个 sink 数据的合并这并不能解决我们的问题我们需要的是跨 checkpoint 的文件合并。
团队考虑过以 row format (e.g. CSV) 输出然后实现自定义的 Hive SerDe使之兼容 RCFile 和 CSV。但很快我们放弃了这个设想因为那样的话需要为每个查询场景实现这个 Hybrid 的 SerDe例如需要为 Presto 实现为 Spark 实现等等。
一方面我们没法投入这么多资源另一方面那种方案也是用户有感的毕竟用户还是需要安装这个自定义的 SerDe。 我们之前提出了生成一个新格式的表但也因为对用户不够透明而被否决。
2. Partition 的可感知性和完整性
如何让下游作业能感知到当天这个 partition 已经 readyactions 表分两级 partition, dt 和 action。action 属于 Hive 的 dynamic partition数量多且不固定。当前 Airflow 下游作业是等待 insert_actions 这个 Hive 任务完成后再开始执行的。这个没问题因为 insert_actions 结束时所有 action 的 partition 都已经 ready 了。但对于 Flink 作业来说没有结束的信号它只能往 Hive 里面提交一个个的 partition如 dt2021-05-29/actionrefresh。因为 action 数量多提交 partition 的过程可能持续数分钟因此我们也不能让 Airflow 作业去感知 dt 级别的 partition那样很可能在只有部分 action 的情况下触发下游。
3. 流式读取云存储文件
项目的输入是不断上传的云存储文件并非来自 MQ (message queue)。Flink 支持 FileStreamingSource可以流式的读入文件但那是基于定时 list 目录以发现新的文件。但这个方案不适合我们的场景因为我们的目录太大云存储 list 操作根本无法完成。
4. Exactly Once 保证
鉴于 actions 表的重要性用户无法接受任何的数据丢失或者重复因此整个方案需要保证恰好一次的处理。
六、整体方案及挑战应对
1. 输出 RCFile 并且避免小文件
我们最终选择的方案是分两步走第一个 Flink 作业以 json (row format) 格式输出然后用另外一个 Flink 作业去做 Json 到 RC 格式的转化。以此解决 Flink 不能愉快的输出合适大小 RC 文件的问题。
输出 json 的中间结果这样我们可以通过 Rolling Policy 控制输出文件的大小可以跨多个 checkpoint 攒成足够大或者时间足够长然后再输出到云存储。这里 Flink 其实利用的是云存储的 Multi Part Upload (MPU) 的功能即每次 checkpoint Flink 也是把当前 checkpoint 攒下来的数据上传至 云存储但输出的不是文件而是一个 part。最后当多个 part 达到大小或者时间要求就可以调用云存储的接口将多个 part 合并成一个文件这个合并操作在云存储端完成应用端无需再次读取这个 part 到本地合并然后再上传。而 Bulk format 均需要一次性全局处理因此无法分段上传然后合并必须一次性全部上传。
当第二个作业感知到一个新的 json 文件上传后加载它转化成 RCFile然后上传到最终的路径。这个过程带来的延迟较小一个文件可以控制在 10s 以内这是可以接受的。
2. 优雅的感知输入文件
输入端没有采用 Flink 的 FileStreamingSource而是采用云存储的 event notification 来感知新文件的产生接受到这个通知后再主动去加载文件。
3. Partition 的可感知性和完整性
输出端我们输出 dt 级别的 success file来让下游可靠地感知日表的 ready。我们实现自定义的 StreamingFileWriter使之输出 partitionCreated 和 partitionInactive 的信号并且通过实现自定义的 PartitionCommitter来基于上述信号判断日表的结束。
其机制如下每个云存储 writer 开始写某个 action会发出一个 partitionCreated 信号当它结束时又发出 partitionInactive 信号。PartitionCommitter 判断某一天之内是否所有的 partittion 都 inactive 了如果是则一天的数据都处理了输出 dt 级别的 success file在 Airflow 通过感知这个文件来判断 Flink 是否完成了日表的处理。
4. Exactly Once
云存储的 event notification 提供 At Least once 保证。Flink 作业内对文件级别进行去重作业采用 Exactly Once 的 checkpoint 设定云存储文件输出基于 MPU 机制等价于支持 truncate因此云存储输出等价于幂等因此等价于端到端的 Exactly Once。
七、项目成果和展望
项目已经上线时延维持在 34 分钟上下其中包括 15 分钟的等待迟到文件。
第一个 Flink 作业需要 8 分钟左右完成 checkpoint 和输出json 转 rc 作业需要 12 分钟完成全部处理。我们可以把这个时间继续压缩但是综合时效性和成本我们选择当前的状态。json 转 rc 作业耗时比当初的预想的要大因为上游作业最后一个 checkpoint 输出太多的文件导致整体耗时长这个可以通过增加作业的并发度线性的下降。输出的文件数比批作业输出的文件数有所增加增加 50% 左右。这是流式处理于批处理的劣势流式处理需要在时间到达时就输出一个文件而此时文件大小未必达到预期。好在这个程度的文件数增加不明显影响下游的性能。做到了下游的完全透明整个上线前后没有收到任何用户异常反馈。
该项目让我们在生产环境验证了利用流式处理框架 Flink 来无缝介入批处理系统实现用户无感的局部改进。将来我们将利用同样的技术去加速更多其他的 Hive 表的生产并且广泛提供更细粒度 Hive 表示的生产例如小时级。另一方面我们将探索利用 data lake 来管理批流一体的数据实现技术栈的逐步收敛。
八、后记
由于采用完全不同的计算框架且需要与批处理系统完全保持一致团队踩过不少的坑限于篇幅无法一一列举。因此我们挑选几个有代表的问题留给读者思考:
为了验证新作业产出的结果与原来 Hive 产出一致我们需要对比两者的输出。那么如何才能高效的比较两个 Hive 表的一致性呢特别是每天有百亿级数据每条有数百个字段当然也包含复杂类型 (array, map, array等)。两个 Flink 作业的 checkpoint 模式都必须是 Exactly Once 吗哪个可以不是哪个必须是?StreamFileWriter 只有在 checkpoint 时才接受到 partitionCreated 和 partitionInactive 信号那么我们可以在它的 snapshotState() 函数里面输出给下游 (下游会保存到 state) 吗?最后一问你们有更好的方案可供我们参考吗?
原文链接 本文为阿里云原创内容未经允许不得转载。
