分享

基于Apache Hudi的流批一体架构实践

本帖最后由 levycui 于 2021-6-29 22:24 编辑
问题导读:
1、模型特征架构是如何演进的?
2、如何设计批流一体平台的构建?
3、Hudi、Delta还是Iceberg如何选择?
4、如何设计Flink构建方案?


1. 前言

当前公司的大数据实时链路如下图,数据源是MySQL数据库,然后通过Binlog Query的方式消费或者直接客户端采集到Kafka,最终通过基于Spark/Flink实现的批流一体计算引擎处理,最后输出到下游对应的存储。

2021-06-29_222747.jpg
2. 模型特征架构的演进
2.1 第一代架构

广告业务发展初期,为了提升策略迭代效率,整理出一套通用的特征生产框架,该框架由三部分组成:特征统计、特征推送和特征获取模型训练。如下图所示:
2021-06-29_222820.jpg

  • 客户端以及服务端数据先通过统一服务Sink到HDFS上
  • 基于基HDFS数据,统计特定维度的总量、分布等统计类特征并推送到Codis中
  • 从Codis中获取特征小时维度模型增量Training,读取HDFS文件进行天级别增量Training

该方案能够满足算法的迭代,但是有以下几个问题

  • 由于Server端直接Put本地文件到HDFS上无法做到根据事件时间精准分区,导致数据源不同存在口径问题
  • 不可控的小文件、空文件问题
  • 数据格式单一,只支持json格式
  • 用户使用成本较高,特征抽取需要不断的Coding
  • 整个架构扩展性较差

为解决上述问题,我们对第一代架构进行了演进和改善,构建了第二代批流一体架构(另外该架构升级也是笔者在饿了么进行架构升级的演进路线)。

2.2 第二代架构
2.2.1 批流一体平台的构建

首先将数据链路改造为实时架构,将Spark Structured Streaming(下文统一简称SS)与Flink SQL语法统一,同时实现与Flink SQL语法大体上一致的批流一体架构,并且做了一些功能上的增强与优化。

2021-06-29_222912.jpg

为什么有了Flink还需要支持SS呢?主要有以下几点原因

  • Spark生态相对更完善,当然现在Flink也做的非常好了
  • 用户使用习惯问题,有些用户对从Spark迁移到Flink没有多大诉求
  • SS Micro Batch引擎的抽象做批流统一更加丝滑
  • 相比Flink纯内存的计算模型,在延迟不敏感的场景Spark更友好

这里举一个例子,比如批流一体引擎SS与Flink分别创建Kafka table并写入到ClickHouse,语法分别如下

Spark Structured Streaming语法如下
  1. --Spark Structured Streaming
  2. CREATE STREAM spark (
  3.     ad_id STRING,
  4.     ts STRING,
  5.     event_ts as to_timestamp(ts)
  6. ) WITH (
  7. 'connector' = 'kafka',
  8. 'topic' = 'xx',
  9. 'properties.bootstrap.servers'='xx',
  10. 'properties.group.id'='xx',
  11. 'startingOffsets'='earliest',
  12. 'eventTimestampField' = 'event_ts',
  13. 'watermark' = '60 seconds',
  14. 'format'='json'
  15. );
  16. create SINK ck(
  17.     ad_id STRING,
  18.     ts STRING,
  19.     event_ts timestamp
  20. ) WITH(
  21. 'connector'='jdbc',
  22. 'url'='jdbc:clickhouse://host:port/db',
  23. 'table-name'='table',
  24. 'username'='user',
  25. 'password'='pass',
  26. 'sink.buffer-flush.max-rows'='10',
  27. 'sink.buffer-flush.interval' = '5s',
  28. 'sink.parallelism' = '3'
  29. 'checkpointLocation'= 'checkpoint_path',
  30. );
  31. insert into ck select * from spark ;
复制代码

Flink SQL语法如下
  1. CREATE TABLE flink (
  2.      ad_id STRING,
  3.      ts STRING,
  4.     event_ts as to_timestamp(ts)
  5.   )
  6. WITH (
  7. 'connector' = 'kafka',
  8. 'topic' = 'xx',
  9. 'properties.bootstrap.servers'='xx',
  10. 'properties.group.id'='xx',
  11. 'scan.topic-partition-discovery.interval'='300s',
  12. 'format' = 'json'
  13. );
  14. CREATE TABLE ck (
  15.     ad_id VARCHAR,
  16.     ts VARCHAR,
  17.     event_ts timestamp(3)
  18.     PRIMARY KEY (ad_id) NOT ENFORCED
  19. ) WITH (
  20. 'connector'='jdbc',
  21. 'url'='jdbc:clickhouse://host:port/db',
  22. 'table-name'='table',
  23. 'username'='user',
  24. 'password'='pass',
  25. 'sink.buffer-flush.max-rows'='10',
  26. 'sink.buffer-flush.interval' = '5s',
  27. 'sink.parallelism' = '3'
  28. );
  29. insert into ck select * from flink ;
复制代码


2.2.2 模型特征处理新架构

新的模型特征处理采用批流一体的架构,上游对接数据源还是Kafka,模型主要有两个诉求
  • 支持增量读取方式减少模型更新的实效性
  • 利用CDC来实现特征的回补

整个流程如下图

2021-06-29_222951.jpg

2.2.3 Hudi、Delta还是Iceberg

3个项目都是目前活跃的开源数据湖方案,feature to feature的展开详细说篇幅太长,大致列举一下各自的优缺点。

2021-06-29_223021.jpg

其实通过对比可以发现各有优缺点,但往往会因为诉求不同,在实际落地生产时3种选型会存在同时多个共存的情况,为什么我们在模型特征的场景最终选择了Hudi呢?主要有以下几点
  • 国内Hudi社区非常活跃,问题可以很快得到解决
  • Hudi对Spark2的支持更加友好,公司算法还是Spark2为主
  • 算法希望有增量查询的能力,而增量查询能力是Hudi原生主打的能力,与我们的场景非常匹配
  • Hudi非常适合CDC场景,对CDC场景支持非常完善

2.2.4 方案上线

我们计划用Spark跟Flink双跑,通过数据质量以及资源成本来选择合适的计算引擎。选择的一个case是广告曝光ed流跟用户点击Click流Join之后落地到Hudi,然后算法增量查询抽取特征更新模型。

2.2.4.1 Flink方案

最初我们用的是Flink 1.12.2 + Hudi 0.8.0,但是实际上发现任务跑起来并不顺利,使用master最新代码0.9.0-SNAPSHOT之后任务可以按照预期运行,运行的Flink SQL如下
  1. CREATE TABLE ed (
  2.     `value` VARCHAR,
  3.     ts as get_json_object(`value`,'$.ts'),
  4.     event_ts as to_timestamp(ts),
  5.     WATERMARK FOR event_ts AS event_ts - interval '1' MINUTE,
  6.     proctime AS PROCTIME()
  7. )WITH (
  8. 'connector' = 'kafka',
  9. 'topic' = 'ed',
  10. 'scan.startup.mode' = 'group-offsets',
  11. 'properties.bootstrap.servers'='xx',
  12. 'properties.group.id'='xx',
  13. 'scan.topic-partition-discovery.interval'='100s',
  14. 'scan.startup.mode'='group-offsets',
  15. 'format'='schemaless'
  16. );
  17. CREATE TABLE click (
  18.     req_id VARCHAR,
  19.     ad_id VARCHAR,
  20.     ts VARCHAR,
  21.     event_ts as to_timestamp(ts),
  22.     WATERMARK FOR event_ts AS event_ts - interval '1' MINUTE,
  23.     proctime AS PROCTIME()
  24. )WITH (
  25. 'connector' = 'kafka',
  26. 'topic' = 'click',
  27. 'properties.bootstrap.servers'='xx',
  28. 'scan.startup.mode' = 'group-offsets',
  29. 'properties.bootstrap.servers'='xx',
  30. 'properties.group.id'='xx',
  31. 'scan.topic-partition-discovery.interval'='100s',
  32. 'format'='json'
  33. );
  34. CREATE TABLE hudi(
  35. uuid VARCHAR,
  36. ts  VARCHAR,
  37. json_info  VARCHAR,  
  38. is_click INT,
  39. dt VARCHAR,
  40. `hour`  VARCHAR,
  41. PRIMARY KEY (uuid) NOT ENFORCED
  42. )
  43. PARTITIONED BY (dt,`hour`)
  44. WITH (
  45.   'connector' = 'hudi',
  46.   'path' = 'hdfs:///xx',
  47.   'write.tasks' = '10',  
  48.   'write.precombine.field'='ts',
  49.   'compaction.tasks' = '1',
  50.   'table.type' = 'COPY_ON_WRITE'  
  51. );
  52. insert into hudi
  53.   SELECT concat(req_id, ad_id) uuid,
  54.   date_format(event_ts,'yyyyMMdd') AS  dt,
  55.   date_format(event_ts,'HH') `hour`,
  56.   concat(ts, '.', cast(is_click AS STRING)) AS ts,
  57.   json_info,is_click
  58. FROM (
  59. SELECT
  60.   t1.req_id,t1.ad_id,t1.ts,t1.json_info,
  61.   if(t2.req_id <> t1.req_id,0,1) as is_click,
  62.   ROW_NUMBER() OVER (PARTITION BY t1.req_id,t1.ad_id,t1.ts ORDER BY if(t2.req_id <> t1.req_id,0,1) DESC) as row_num
  63.   FROM
  64.   (select  ts,event_ts,map_info['req_id'] req_id,map_info['ad_id'] ad_id, `value` as json_info from ed,LATERAL TABLE(json_tuple(`value`,'req_id','ad_id')) as T(map_info)) t1  
  65.   LEFT JOIN
  66.   click t2
  67.   ON t1.req_id=t1.req_id and t1.ad_id=t2.ad_id
  68.   and t2.event_ts between t1.event_ts - INTERVAL '10' MINUTE and t1.event_ts + INTERVAL '4' MINUTE
  69.   ) a where a.row_num=1;
复制代码



标注:上述SQL中有几处与官方SQL不一致,主要是实现了统一规范Schema为一列的Schemaless的Format、与Spark/Hive语义基本一致的get_json_object以及json_tuple UDF,这些都是在批流一体引擎做的功能增强的一小部分。

但是在运行一周后,面临着业务上线Delay的压力以及暴露出来的两个问题让我们不得不先暂时放弃Flink方案

  • 任务反压的问题(无论如何去调整资源似乎都会出现严重的反压,虽然最终我们通过在写入Hudi之前增加一个upsert-kafka的中间流程解决了,但链路过长这并不是我们预期内的)
  • 还有一点是任务存在丢数据的风险,对比Spark方案发现Flink会有丢数据的风险

标注:这个case并非Flink集成Hudi不够,国内已经有很多使用Flink引擎写入Hudi的实践,但在我们场景下因为为了确保上线时间,没有太多时间细致排查问题。实际上我们这边Kafka -> Hive链路有95%的任务都使用Flink替代了Spark Structured Streaming(SS)

2.2.4.2 Spark方案

由于没有在Hudi官方网站上找到SS集成的说明,一开始笔者快速实现了SS与Hudi的集成,但是在通读Hudi代码之后发现其实社区早已有了SS的完整实现,另外咨询社区同学leesf之后给出的反馈是当前SS的实现也很稳定。稍作适配SS版本的任务也在一天之内上线了,任务SQL如下

  1. CREATE STREAM ed (
  2.     value STRING,
  3.     ts as get_json_object(value,'$.ts'),
  4.     event_ts as to_timestamp(get_json_object(value,'$.ts'))
  5. ) WITH (
  6. 'connector' = 'kafka',
  7. 'topic' = 'ed',
  8. 'properties.bootstrap.servers'='xx',
  9. 'properties.group.id'='xx',
  10. 'startingOffsets'='earliest',
  11. 'minPartitions' = '60',
  12. 'eventTimestampField' = 'event_ts',
  13. 'maxOffsetsPerTrigger' = '250000',   
  14. 'watermark' = '60 seconds',
  15. 'format'='schemaless'
  16. );
  17. CREATE STREAM  click (
  18.     req_id STRING,
  19.     ad_id STRING,
  20.     ts STRING,
  21.     event_ts as to_timestamp(ts)
  22. ) WITH (
  23. 'connector' = 'kafka',
  24. 'topic' = 'click',
  25. 'properties.bootstrap.servers'='xxxx'properties.group.id'='dw_ad_algo_naga_dsp_ed_click_rt',
  26. 'startingOffsets'='earliest',
  27. 'maxOffsetsPerTrigger' = '250000',
  28. 'eventTimestampField' = 'event_ts',
  29. 'minPartitions' = '60',
  30. 'watermark' = '60 seconds',
  31. 'format'='json'
  32. );
  33. --可以动态注册python、java、scala udf
  34. create python function py_f with (
  35. 'code' = '
  36. def apply(self,m):
  37.   return 'python_{}'.format(m)
  38. ',
  39. 'methodName'= 'apply',
  40. 'dataType' = 'string'
  41. );
  42. create SINK hudi(
  43. uuid STRING,
  44. dt STRING,
  45. hour  STRING,
  46. ts  STRING,
  47. json_info  STRING,  
  48. is_click INT
  49. ) WITH (
  50.     'connector'='hudi',
  51.     'hoodie.table.name' = 'ed_click',
  52.     'path' ='hdfs:///xx',
  53.     'hoodie.datasource.write.recordkey.field' = 'uuid',
  54.     'hoodie.datasource.write.precombine.field' = 'ts',
  55.     'hoodie.datasource.write.operation' = 'upsert',
  56.     'hoodie.datasource.write.partitionpath.field' = 'dt,hour',
  57.     'hoodie.datasource.write.keygenerator.class'= 'org.apache.hudi.keygen.ComplexKeyGenerator',
  58.     'hoodie.datasource.write.table.type' = 'COPY_ON_WRITE',
  59.     'hoodie.datasource.write.hive_style_partitioning'='true',
  60.     'hoodie.datasource.write.streaming.ignore.failed.batch'='false',
  61.     'hoodie.keep.min.commits'='120',
  62.     'hoodie.keep.max.commits'='180',
  63.     'hoodie.cleaner.commits.retained'='100',
  64.     --'hoodie.datasource.write.insert.drop.duplicates' = 'true',
  65.     --'hoodie.fail.on.timeline.archiving'='false',
  66.     --'hoodie.datasource.hive_sync.table'='true',
  67.    -- 'hoodie.datasource.hive_sync.database'='ods_test',
  68.    -- 'hoodie.datasource.hive_sync.table'='ods_test_hudi_test2',
  69.    -- 'hoodie.datasource.hive_sync.use_jdbc'='false',
  70.    -- 'hoodie.datasource.meta.sync.enable' ='true',
  71.    -- 'hoodie.datasource.hive_sync.partition_fields'='dt,hour',
  72.    -- 'hoodie.datasource.hive_sync.partition_extractor_class'='org.apache.hudi.hive.MultiPartKeysValueExtractor',
  73.     'trigger'='30',
  74.     'checkpointLocation'= 'checkpoint_path'
  75. );
  76. INSERT INTO
  77.    hudi
  78. SELECT
  79.   concat(req_id, ad_id) uuid,
  80.   date_format(ts,'yyyyMMdd') dt,
  81.   date_format(ts,'HH') hour,
  82.   concat(ts, '.', cast(is_click AS STRING)) AS ts,
  83.   json_info,
  84.   is_click
  85. FROM
  86.   (
  87.     SELECT
  88.       t1.req_id,
  89.       t1.ad_id,
  90.       t1.ts,
  91.       t1.json_info,
  92.       IF(t2.req_id is null, 0, 1) AS is_click
  93.     FROM
  94.       (select  ts,event_ts,req_id,ad_id,value as json_info from ed
  95.       lateral view json_tuple(value,'req_id','ad_id') tt as req_id,ad_id) t1
  96.       LEFT JOIN click t2 ON t1.req_id = t2.req_id
  97.       AND t1.ad_id = t2.ad_id
  98.       AND t2.event_ts BETWEEN t1.event_ts - INTERVAL 10 MINUTE
  99.       AND t1.event_ts + INTERVAL 4 MINUTE
  100.   ) tmp;
复制代码

标注:Spark批流一体引擎在流语法上尽量与Flink对齐,同时我们实现了python/java/scala多语言udf的动态注册以方便用户使用

3. 新方案收益

通过链路架构升级,基于Flink/Spark + Hudi的新的流批一体架构带来了如下收益

  • 构建在Hudi上的批流统一架构纯SQL化极大的加速了用户的开发效率
  • Hudi在COW以及MOR不同场景的优化让用户有了更多的读取方式选择,增量查询让算法可以实现分钟级别的模型更新,这也是用户的强烈诉求
  • 利用SS以及Flink的事件时间语义抹平了口径上的Gap
  • Hudi自动Compact机制+小文件智能处理,对比第一版实现甚至对比需要手动Compact无疑极大的减轻了工程负担

4. 踩过的坑

  • 写Hudi重试失败导致数据丢失风险。解决办法:hoodie.datasource.write.streaming.ignore.failed.batch设置为false,不然Task会间隔hoodie.datasource.write.streaming.retry.interval.ms(默认2000)重试hoodie.datasource.write.streaming.retry.count(默认3)
  • 增量查询Range太大,导致算法任务重试1小时之前的数据获取到空数据。解决办法:调大保留版本数对应参数为hoodie.keep.min.commits、hoodie.keep.max.commits调大cleanup retention版本数对应参数为hoodie.cleaner.commits.retained
  • Upsert模式下数据丢失问题。解决办法:hoodie.datasource.write.insert.drop.duplicates设置为false,这个参数会将已经存在index的record丢弃,如果存在update的record会被丢弃
  • Spark读取hudi可能会存在path not exists的问题,这个是由于cleanup导致的,解决办法:调整文件版本并进行重试读取

5. 未来规划

基于Hudi线上运行的稳定性,我们也打算基于Hudi进一步探索流批一体的更多应用场景,包括

使用Hudi替代Kafka作为CDC实时数仓Pipeline载体
  • 深度结合Hive以及Presto,将Hive表迁移为基于Hudi的架构,以解决分区小文件以及产出失效的问题
  • 探索Flink+Hudi作为MySQL Binlog归档方案
  • 探索Z-Order加速Spark在多维查询上的性能表现

作者:易伟平
来源:https://mp.weixin.qq.com/s/LPcWr-o1KWVa-xpAiOwjig


最新经典文章,欢迎关注公众号



已有(1)人评论

跳转到指定楼层
若无梦何远方 发表于 2021-6-30 21:06:48
坚持每日一读
回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条