谈到数据仓库， 肯定离不开应用Extract-Transform-Load (ETL)或 Extract-Load-Transform (ELT)。 将起源不同、格局各异的数据提取到数据仓库中，并进行解决加工。传统的数据转换过程个别采纳Extract-Transform-Load (ETL)来将业务数据转换为适宜数仓的数据模型，然而，这依赖于独立于数仓外的ETL零碎，因此保护老本较高。

ByConity 作为云原生数据仓库，从0.2.0版本开始逐渐反对 Extract-Load-Transform (ELT)，使用户免于保护多套异构数据系统。本文将介绍 ByConity 在ELT方面的能力布局，实现原理和应用形式等。

ETL场景和计划

ELT与ETL的区别

  • ETL：是用来形容将数据从起源端通过抽取、转置、加载至目标端（数据仓库）的过程。Transform通常形容在数据仓库中的前置数据加工过程。

  • ELT 专一于将最小解决的数据加载到数据仓库中，而把大部分的转换操作留给分析阶段。相比起前者（ETL)，它不须要过多的数据建模，而给剖析者提供更灵便的选项。ELT曾经成为当今大数据的解决常态，它对数据仓库也提出了很多新的要求。

资源重复的挑战

典型的数据链路如下：咱们将行为数据、日志、点击流等通过MQ/ Kafka/ Flink将其接入存储系统当中，存储系统又可分为域内的HDFS 和云上的 OSS& S3 这种近程贮存零碎，而后进行一系列的数仓的ETL操作，提供给 OLAP零碎实现剖析查问。

但有些业务须要从上述的存储中做一个分支，因而会在数据分析的某一阶段，从整体链路中将数据导出，做一些不同于主链路的ETL操作，会呈现两份数据存储。其次在这过程中也会呈现两套不同的ETL逻辑。

当数据质变大，计算冗余以及存储冗余所带来的老本压力也会愈发变大，同时，存储空间的收缩也会让弹性扩容变得不便当。

业界解决思路

在业界中，为了解决以上问题，有以下几类流派：

  • 数据预计算流派：如Kylin等。如果Hadoop零碎中出报表较慢或聚合能力较差，能够去做一个数据的预计算，提前将配的指标的cube或一些视图算好。理论SQL查问时，能够间接用外面的cube或视图做替换，之后间接返回。
  • 流批一体 派：如 Flink、Risingwave。在数据流进时，针对一些须要出报表或者须要做大屏的数据间接内存中做聚合。聚合实现后，将后果写入HBase或MySQL中再去取数据，将数据取出后作展现。Flink还会去间接裸露中间状态的接口，即queryable state，让用户更好的应用状态数据。然而最初还会与批计算的后果实现对数，如果不统一，须要进行回查操作，整个过程考验运维/开发同学的功力。
  • 湖仓 一体&HxxP:将数据湖与数据仓库联合起来。

ELT in ByConity

整体执行流程

ELT工作对系统的要求：

  1. 整体易扩大：导入和转换通常须要大量的资源，零碎须要通过程度扩大的形式来满足数据量的快速增长。
  2. 可靠性和容错能力：大量的job能有序调度；呈现task偶尔失败（OOM）、container失败时，可能拉起重试；能解决肯定的数据歪斜
  3. 效率&性能：无效利用多核多机并发能力；数据疾速导入；内存应用无效（内存治理）；CPU优化（向量化、codegen）
  4. 生态& 可观测性：可对接多种工具；工作状态感知；工作进度感知；失败日志查问；有肯定可视化能力

ByConity 针对ELT工作的要求，以及以后场景遇到的艰难，新增了以下个性和优化改良。

分阶段执行（Stage-level Scheduling）


原理解析

  • 以后 ClickHouse的 SQL 执行过程如下：

    • 第一阶段，Coordinator 收到分布式表查问后将申请转换为对 local 表查问发送给每个 shard 节点；
    • 第二阶段，Coordinator 收到各个节点的后果后汇聚起来解决后返回给客户端；

  • ClickHouse 将Join操作中的右表转换为子查问，带来如下几个问题都很难以解决：

    • 简单的query有多个子查问，转换复杂度高；
    • Join表较大时，容易造成worker节点的OOM；
    • 聚合阶段在Cooridnator，压力大，容易成为性能瓶颈；



不同于ClickHouse，咱们在ByConity 中实现了对简单查问的执行优化。通过对执行打算的切分，将之前的两阶段执行模型转换为分阶段执行。在逻辑打算阶段，依据算子类型插入exchange算子。执行阶段依据exchange算子将整个执行打算进行DAG切分，并且分stage进行调度。stage之间的exchange算子负责实现数据传输和替换。

要害节点：

  1. exchange节点插入
  2. 切分stage
  3. stage scheduler
  4. segment executer
  5. exchange manager

这里重点来讲一下exchange的眼帘。上图能够看到，最顶层的是query plan。上面转换成物理打算的时候，咱们会依据不同的数据分布的要求转换成不同的算子。source层是接收数据的节点，根本都是对立的，叫做ExchangeSource。Sink则有不同的实现，BroadcastSink、Local、PartitionSink等，他们是作为map task的一部分去运行的。如果是跨节点的数据操作，咱们在底层应用对立的brpc流式数据传输，如果是本地，则应用内存队列来实现。针对不同的点，咱们进行了十分粗疏的优化：

  • 数据传输层

    • 过程内通过内存队列，无序列化，zero copy
    • 过程间应用brpc stream rpc，保序、连贯复用、状态码传输、压缩等

  • 算子层

    • 批量发送
    • 线程复用，缩小线程数量

带来的收益

因为ByConity 彻底采纳了多阶段的查问执行形式，整体有很大的收益：

  • Cooridnator更稳固、更高效

    • 聚合等算子拆分到worker节点执行
    • Cooridnator节点只须要聚合最终后果

  • Worker OOM缩小

    • 进行了stage切分，每个stage的计算绝对简略
    • 减少了exchange算子，缩小内存压力

  • 网络连接更加稳固、高效

    • exchange算子无效传输
    • 复用连接池

自适应的调度器（Adaptive Scheduler）

Adaptive Scheduler 属于咱们在稳定性方面所做的个性。在OLAP场景中可能会发现局部数据不全或数据查问超时等，起因是每个worker是所有的query共用的，这样一旦有一个worker 较慢就会导致整个query的执行受到影响。

计算节点共用存在的问题：

  • Scan 所在的节点负载和不同查问所需的扫描数据量相干，做不到齐全均匀；
  • 各 Plan Segment 所需资源差别大；

这就导致worker节点之间的负载重大不平衡。负载较重的worker节点就会影响query整体的过程。因而咱们做了以下的优化计划：

  • 建设 Worker 衰弱度机制。Server 端建设 Worker 衰弱度治理类，能够疾速获取 Worker Group 的衰弱度信息，包含CPU、内存、运行Query数量等信息。
  • 自适应调度：每个SQL 依据 Worker 衰弱度动静的进行抉择以及计算节点并发度管制。

查问的队列机制（Query Queue）


咱们的集群也会呈现满载状况，即所有的worker都是不衰弱的或者满载/超载的，就会用查问队列来进行优化。

咱们间接在server端做了一个manager。每次查问的时候manager会去check集群的资源，并且持有一个锁。如果资源不够用，则期待资源开释后去唤醒这个锁。这就防止了Server端不限度的下发计算工作，导致worker节点超载，而后崩掉的状况。

以后实现绝对简略。server是多实例，每个server实例中都有queue，所持有的是一个部分视角，不足全局的资源视角。除此之外，每个queue中的查问状态没有长久化，只是简略的缓存在内存中。

后续，咱们会减少server之间的协调，在一个全局的视角上对查问并发做限度。也会对server实例中query做长久化，减少一些failover的场景反对。

异步执行（Async Execution）

ELT工作的一个典型特色就是：绝对于即时剖析，他们的运行工夫会绝对较长。个别ELT工作执行时长为分钟级，甚至达到小时级。

目前 ClickHouse的客户端查问都采纳阻塞的形式进行返回。这样就造成了客户端长期处于期待的状况，而在这个期待过程中还须要放弃和服务端的连贯。在不稳固的网络状况下，客户端和服务端的连贯会断开，从而导致服务端的工作失败。

为了缩小这种不必要的失败，以及缩小客户端为了维持连贯的减少的复杂度。咱们开发了异步执行的性能，它的实现如下：

  1. 用户指定异步执行。用户能够通过settings enable_async_query = 1的形式进行per query的指定。也能够通过set enable_async_query = 1的形式进行session级别的指定。
  2. 如果是异步query，则将其放到后盾线程池中运行
  3. 静默io。当异步query执行时，则须要切断它和客户端的交互逻辑，比方输入日志等。

针对query的初始化还是在session的同步线程中进行。一旦实现初始化，则将query状态写入到metastore，并向客户端返回async query id。客户端能够用这个id查问query的状态。async query id返回后，则示意实现此次查问的交互。这种模式下，如果语句是select，那么后续后果则无奈回传给客户端。这种状况下咱们举荐用户应用async query + select...into outfile的组合来满足需要。

将来布局

针对ELT混合负载，ByConity 0.2.0版本目前只是牛刀小试。后续的版本中咱们会继续优化查问相干的能力，ELT为外围的布局如下：

故障恢复能力

  • 算子Spill

    • Sort、Agg、Join 算子Spill；
    • Exchange Spill 能力；

  • Recoverability 容错复原

    • 算子执行复原：ELT工作运行时长较长时，两头 Task的偶发失败会导致整个Query失败，反对Task 级别重试能够极大地升高环境起因导致的偶发失败；
    • Stage重试：当节点失败时，能够进行 Stage级别的重试；
    • 保留队列作业状态的能力；
  • Remote Shuffle Service：以后业界开源的 shuffle service通常为Spark定制，没有通用的客户端，比方c++客户端。后续咱们会补充这部分能力。

资源

  • 计算资源可指定：用户可指定query须要的计算资源；
  • 计算资源预估/预占：可动静预估query须要的计算资源，并通过预占的形式进行调配；
  • 动静申请资源：以后worker均为常驻过程/节点。动静申请资源能够进步利用率；
  • 更细粒度的资源隔离：通过worker group或者过程级别的隔离，缩小各query之间相互影响；

欢送退出社区，与咱们共建

ByConity 我的项目 GitHub 地址：

https://github.com/ByConity

用户手册：

https://byconity.github.io/zh-cn/docs/introduction/background...

扫码增加小助手