背景介绍
Linkis是一款优秀的计算中间件,他对应用层屏蔽了复杂的底层计算引擎和存储方案,让大数据变得更加简单易用,同时也让运维变得更加方便。我们的平台很早就部署了WDS全家桶给业务用户和数据分析用户使用。近段时间,我们也调研和实现了hudi作为我们数据湖落地的方案,他帮助我们解决了在hdfs上进行实时upsert的问题,让我们能够完成诸如实时ETL,实时对账等项目。hudi作为一个数据湖的实现,我觉得他也是一种数据存储方案,所以我也希望它能够由Linkis来进行管理,这样我们的平台就可以统一起来对外提供能力。因此我这边做了一个Linkis和Hudi的结合和使用的分享。
组件 | 版本 |
---|---|
EMR | 6.2.0 |
hadoop | 3.2.1-amzn |
spark | 3.0.1-amzn |
flink | 1.13.1 |
hive | 3.1.2-amzn |
java | openJDK1.8 |
scala | 2.12.10 |
linkis | 1.1.3 |
DSS | 1.1.0 |
hudi | 0.10.1 |
mvn clean package -Dmaven.test.skip=true -Dspark3.0.x -Dscala-2.12 -Dcheckstyle.skip=true -Drat.skip=true
根据架构图所示,我们可以看到,业务库的binlog可以通过CDC直接到Hudi或者先经过Kafka再到Hudi。CDC和KafkaToHudi的应用使用Streamis进行提交,Streamis再通过Linkis将任务提交给Flink执行,这样用户的业务表就可以近实时地同步到我们的hudi表中。用户使用DSS进行查询,也是通过Linkis的Spark引擎访问hudi。
Spark引擎引入hudi的话,我们一般推荐直接将hudi-spark-bundle.jar放到${SPARK_HOME}/jars中,同时修改spark-defaut.conf,添加下面的配置
spark.serializer org.apache.spark.serializer.KryoSerializer
spark.sql.extensions org.apache.spark.sql.hudi.HoodieSparkSessionExtension
同时也可以调大以下参数用来加大序列化的buffer内存。
spark.kryoserializer.buffer.max 256m
spark.kryoserializer.buffer 256k
当然,也可以修改linkis启动spark引擎的方式,判断如果用户如果需要hudi的读取,就通过 --jars的方式引入hudi-spark-bundle.jar,并通过--conf的方式修改spark的序列化器参数。
Flink引擎的话,就比较简单了,直接将hudi-flink-bundle.jar放置在${FLINK_HOME}/lib目录下即可,经过测试,hudi-flink-bundle.jar有shaded的方式,不会引入和其他connector冲突的包。
将hudi引入到Linkis之后,我们可以直接通过streamis编写实时ETL任务,将业务表近实时地落到hudi,用户看到的最新的数据将是分钟级别的最新数据,而不是t-1或者几小时前的数据。而且这个实时ETL对集群压力也不大,如果是spark跑批,跑一天的数据,将会占据很大的集群资源,而且资源就在那一段时间被占用。
我们以前的对账是在tidb上面,进行执行,tidb运维难度较大,而且商业版本价格较高,我们直接将数据导入到hudi之后,使用spark进行计算对账,也能达到分钟级别的延迟,同时运维成本降低。
实时BI也是hudi的一个应用,通过Linkis的presto引擎查询hudi表,可以在visualis或tableau中实时刷新报表。presto的配置可以查看presto与hudi的连接。
用户通过DSS直接查询hudi表,来进行取数以及实时分析,可以更快地反映出当天时刻的业务状况。