更多技术交流、求职机会,欢迎关注字节跳动数据平台微信公众号,回复【1】进入官方交流群
谈到数据仓库, 一定离不开使用Extract-Transform-Load (ETL)或 Extract-Load-Transform (ELT)。 将来源不同、格式各异的数据提取到数据仓库中,并进行处理加工。
传统的数据转换过程一般采用Extract-Transform-Load (ETL)来将业务数据转换为适合数仓的数据模型,然而,这依赖于独立于数仓外的ETL系统,因而维护成本较高。现在,以火山引擎ByteHouse为例的云原生数据仓库,凭借其强大的计算能力、可扩展性,开始全面支持Extract-Load-Transform (ELT)的能力,从而使用户免于维护多套异构系统。具体而言,用户可以将数据导入后,通过自定义的SQL语句,在ByteHouse内部进行数据转换,而无需依赖独立的ETL系统及资源。
火山引擎ByteHouse是一款基于开源ClickHouse推出的云原生数据仓库,本篇文章将介绍ByteHouse团队如何在ClickHouse的基础上,构建并优化ELT能力,具体包括四部分:ByteHouse在字节的应用、ByteHouse团队做ELT的初衷、ELT in ByteHouse实现方案、未来规划。
从2017年开始,字节内部的整体数据量不断上涨,为了支撑实时分析的业务,字节内部开始了对各种数据库的选型。经过多次实验,在实时分析版块,字节内部决定开始试水ClickHouse。
2018年到2019年,字节内部的ClickHouse业务从单一业务,逐步发展到了多个不同业务,适用到更多的场景,包括BI 分析、A/B测试、模型预估等。
在上述这些业务场景的不断实践之下,研发团队基于原生ClickHouse做了大量的优化,同时又开发了非常多的特性。
2020年, ByteHouse正式在字节跳动内部立项,2021年通过火山引擎对外服务。
截止2022年3月,ByteHouse在字节内部总节点数达到18000个,而单一集群的最大规模是2400个节点。
在火山引擎官网的产品页中,我们可以搜到ByteHouse产品(如下图):
ByteHouse产品可以分为两个形态:
在数据量较小、使用较为简单的情况下,用户可以先试用企业版本,如果之后集群规模变大、运维压力较大,亦或是扩展能力要求变高,那么就可以转用到纯算分离、运维能力更强的CDW上来,也就是我们刚刚提及的数仓版。
数据洞察是支持千亿级别数据自助分析的一站式数据分析及协作平台,包括数据导入以及整合查询分析,最终以数据门户、数字大屏、管理驾驶舱的可视化形态呈现给业务用户,为一个比较典型的场景。
用户行为分析,即多场景决策的数据分析平台。而在增长分析当中,分为了以下三个内容:
以懂车帝为例,懂车帝主要给用户提供真实、专业汽车的内容分享和高效的选车服务,同时基于营销需求,他们会根据用户增长的模型以及销售方法论,收集用户在端内的操作行为,进行后台的查询分析。
而这种查询分析底层对接了ByteHouse的大数据引擎,最后实现秒级甚至是亚秒级分析的决策。整个过程包括智能诊断、智能规划以及策略到投放效果评估闭环,最终实现智能营销和精细化运营。
下面表述上会有一些两个词语混用的场景,大家不必过分关注区别。
传统大数据解决的方案有两大难点:慢和难。分别体现在传统大数据方案在及时性上达不到要求以及传统数仓ETL对人员要求高、定位难和链路复杂。
但是ByteHouse可以轻松的解决上述问题:将hive数据直接导入到ByteHouse,形成大宽表,后续所有处理都在ByteHouse进行。
典型的数据链路如下:我们将行为数据、日志、点击流等通过MQ/Kafka/Flink将其接入存储系统当中,存储系统又可分为域内的HDFS和云上的OSS&S3这种远程储存系统,然后进行一系列的数仓的ETL操作,提供给OLAP系统完成分析查询。
但有些业务需要从上述的存储中做一个分支,因此会在数据分析的某一阶段,从整体链路中将数据导出,做一些不同于主链路的ETL操作,会出现两份数据存储。其次在这过程中也会出现两套不同的ETL逻辑。
当数据量变大,计算冗余以及存储冗余所带来的成本压力也会愈发变大,同时,存储空间的膨胀也会让弹性扩容变得不便利。
从OLAP场景扩展出去,随着数据量的增长和业务复杂度的提升,ClickHouse渐渐不能满足要求,体现在以下几点:
在ByteHouse上去做复杂查询或ELT任务,可以扩展ClickHouse的能力,增强它的可用性、稳定性以及性能,同时还支持不同类型的混合负载。
在业界中,为了解决以上问题,有以下几类流派:
ELT任务对系统的要求:
ByteHouse针对ELT任务的要求,以及当前场景遇到的困难,做了如下特性和改进。
方案:
val df = spark.read.format("CnchPart")
.options(Map("table" -> "cnch_db.c1")).load()
val spark = SparkSession.builder()
.appName("CNCH-Reader")
.config("spark.sql.extensions", "CnchAutoConvertExtension")
.enableHiveSupport() .getOrCreate()
val df = spark.sql("select * from cnch_db.c1")
收益:
不同于ClickHouse,我们在ByteHouse中实现了对复杂查询的执行优化。通过对执行计划的切分,将之前的两阶段执行模型转换为分阶段执行。在逻辑计划阶段,根据算子类型插入exchange算子。执行阶段根据exchange算子将整个执行计划进行DAG切分,并且分stage进行调度。stage之间的exchange算子负责完成数据传输和交换。
关键点:
这里重点来讲一下exchange的视线。上图可以看到,最顶层的是query plan。下面转换成物理计划的时候,我们会根据不同的数据分布的要求转换成不同的算子。source层是接收数据的节点,基本都是统一的,叫做ExchangeSource。Sink则有不同的实现,BroadcastSink、Local、PartitionSink等,他们是作为map task的一部分去运行的。如果是跨节点的数据操作,我们在底层使用统一的brpc流式数据传输,如果是本地,则使用内存队列来实现。针对不同的点,我们进行了非常细致的优化。
这是在稳定性方面所做的特性。在OLAP场景中可能会发现部分数据不全或数据查询超时等,原因是每个计算节点是所有的query共用的,这样一旦有一个节点较慢就会导致整个query的执行受到影响。
计算节点共用存在的问题:
这就导致worker节点之间的负载严重不均衡。负载较重的worker节点就会影响query整体的进程。
解决措施:
我们的集群也会出现满载情况,即所有的worker都是不健康的或者满载/超载的,就会用查询队列来进行优化。
我们直接在server端做了一个manager。每次查询的时候manager会去check集群的资源,并且持有一个锁。如果资源不够用,则等待资源释放后去唤醒这个锁。这就避免了Server端不限制的下发计算任务,导致worker节点超载,然后崩掉的情况。
当前实现相对简单。server是多实例,每个server实例中都有queue,所持有的是一个局部视角,缺乏全局的资源视角。除此之外,每个queue中的查询状态没有持久化,只是简单的缓存在内存中。
后续,我们会增加server之间的协调,在一个全局的视角上对查询并发做限制。也会对server实例中query做持久化,增加一些failover的场景支持。
ELT任务的一个典型特征就是相对即时分析,他们的运行时间会相对较长。一般为分钟级,甚至到达小时级。目前ClickHouse的客户端查询都采用阻塞的方式进行返回。这样就造成了客户端长期处于等待的情况,而在这个等待过程中还需要保持和服务端的连接。在不稳定的网络情况下,客户端和服务端的连接会断开,从而导致服务端的任务失败。
为了减少这种不必要的失败,以及减少客户端为了维持连接的增加的复杂度。我们开发了异步执行的功能,它的实现如下:
针对query的初始化还是在session的同步线程中进行。一旦完成初始化,则将query状态写入到metastore,并向客户端返回async query id。客户端可以用这个id查询query的状态。async query id返回后,则表示完成此次查询的交互。这种模式下,如果语句是select,那么后续结果则无法回传给客户端。这种情况下我们推荐用户使用async query + select...into outfile的组合来满足需求。
针对ELT混合负载,目前只是牛刀小试。后续的版本中我们会持续补齐规划中的能力,包括但不限于以下:
点击跳转「链接」了解更多
更新时间:2024-09-12
本站资料均由网友自行发布提供,仅用于学习交流。如有版权问题,请与我联系,QQ:4156828
© CopyRight 2008-2024 All Rights Reserved. Powered By bs178.com 闽ICP备11008920号-3
闽公网安备35020302034844号