一、企业数据化系统建设阶段历程：
随着技术的飞速发展，企业对数据的需求不断增长，数据已然成为企业的宝贵资产之一，但我们不仅要收集和存储数据，更要挖掘和利用数据。数据仓库已经成为了企业信息化建设的重要组成部分。将企业内部各个部门的数据整合在一起，形成一个统一的数据视图。这样，企业决策者可以更加方便地获取所需的信息，从而提高决策效率。
数据仓库系统的作用能实现跨业务条线、跨系统的数据整合，为管理分析和业务决策提供统一的数据支持。数据仓库能够从根本上帮助你把公司的运营数据转化成为高价值的可以获取的信息（或知识），并且在恰当的时候通过恰当的方式把恰当的信息传递给恰当的人。
但是，在实际业务中，用户会基于不同的产品业务线分别构建实时数仓和离线数仓，用来满足日常企业的高效的查询和分析功能，帮助企业更快地找到问题的根源，从而降低运营成本。
今天带大家来了解一款由字节跳动开源云原生数据仓库 – “ByConity”，可以满足用户的多种数据分析场景，基于 ClickHouse 进行优化和扩展，擅长交互式查询和即席查询，具有支持多表关联复杂查询、集群扩容无感、离线批数据和实时数据流统一汇总等特点。同时，ByConity 增加了 bsp 模式：
二、ByConity 产品介绍：
ByConity是字节跳动开源的分布式的云原生 SQL 数仓引擎，基于 ClickHouse 进行优化和扩展，擅长交互式查询和即席查询，具有支持多表关联复杂查询、集群扩容无感、离线批数据和实时数据流统一汇总等特点。
ByConity 采用计算存储分离的架构、主流的 OLAP 引擎和自研的表引擎，提供便捷的弹性扩缩容和极速的分析性能，覆盖实时分析和海量数据的离线分析，支持多表关联复杂查询、集群扩容无感、离线批数据和实时数据流统一汇总等功能，特别擅长交互式查询和即席查询，帮助企业更好地挖掘数据价值。
2.1. 技术背景和架构：
ByConity 的设计初衷是为了解决 ClickHouse 在使用过程中遇到的运维成本高、扩缩容成本高以及关联查询局限性等问题。通过重新设计存算分离架构，ByConity 能够更好地适应字节内部快速发展的业务需求，提供优异的读写性能和资源弹性扩缩容能力，ByConity 的架构分为三层，包括服务接入层，计算层和数据存储层。
①. 服务接入层负责客户端数据和服务的接入，也就是 ByConity Server。
②. ByConity 的计算资源层，由一个或者多个计算组构成，每个 Virtual Warehouse（VW）是一个计算组。
③. 数据存储层由分布式文件系统，如 HDFS、S3 等构成。
2.2 ByConity 的主要功能特性：
ByConity 采用存算分离的云原生架构，通过这种架构获得了弹性和降低资源浪费的优势，但与此同时也在一定程度上提高了产品的复杂度。定位为云原生数据仓库，是希望能够承担更多类型、更复杂的分析任务负载，无论是在线的实时分析还是离线数据的清洗/加工任务都能够胜任。更全面的能力能够帮助用户降低数据分析平台的整体复杂度。
①. 计算与存储分离：将计算和存储资源解耦，使得每个计算节点成为一个无状态的单纯计算节点，利用分布式存储的扩展能力和计算节点的无状态特性实现动态的扩缩容。
②. 资源隔离：对不同的租户进行资源的隔离，确保租户之间不会相互影响。
③. 读写分离：确保读操作和写操作不会相互影响，提高系统稳定性。
④. 弹性扩缩容：支持弹性的扩缩容，能够实时、按需地对计算资源进行扩缩容，保证资源的高效利用。
⑤. 数据强一致性：确保数据读写的强一致性，保证数据始终是最新的。
⑥. 高性能：采用列存、向量化执行、MPP 执行、查询优化等技术，提供优异的读写性能。
2.3 部署和运行：
在部署 ByConity 时，需要准备相应的资源，包括操作系统版本为 CentOS 8.2 的服务器，并使用公网 yum 源。硬件规格中，Worker 和 Server 的本地磁盘主要用于存储写入时的临时数据和日志文件，同时 Worker 的本地磁盘还会存储数据的 Cache4。部署过程中需要使用 Kubernetes 命令行工具 kubectl 进行管理和操作，包括创建、更新、删除和查看资源，监控其状态和日志等。
三、测试环境标准测试 TPC-DS 数据集：
社区为了让大家实际上手感受 bsp 模式带来的效果，温馨的提供测试环境与测试集，用户可以在提供好的环境中上手测试，使用小资源跑 TPC-DS 数据集，可以自己在测试过程中增加一些参数进行调整。
测试环境提供了使用 TPC-DS 1TB 的数据测试数据集，可以验证 ByConity 在 ELT 方面的实际感受。通过 SSH 等工具连接集群，并完成基于 TPC-DS 标准数据集的数据查询等操作。
3.1 TPC-DS 1TB 的数据测试数据集说明：
TPC-DS 是一个面向决策支持的基准，它对决策支持系统的几个普遍适用的方面进行建模，包括查询和数据维护等，使大数据系统等新兴技术能够执行基准测试，因为测试环境提供了使用 TPC-DS 1TB 的数据测试数据集。
TPC-DS 1TB 的数据测试数据集包含 7 张事实表，17 张维度表平均每张表含有 18 列,负载包含 99 个 SQL 查询，覆盖大数据集的统计、报表生成、联机查询、数据挖掘等复杂应用。
使用 clickhouse client 将 s3 的 csv 的数据导入到数据库中，为了安全起见，将 id 与 secret 去掉，并且设置 max_execution_time、receive_timeout、send_timeout 等相关参数。
#!/bin/bash
clickhouse client --port 9010 --query="INSERT INTO test_elt.catalog_page FORMAT CSV INFILE 's3://dpsg-dataset/titan/benchmark/tpcds_csv_1000/catalog_page.dat' settings s3_ak_id ='', s3_ak_secret ='', s3_endpoint = 'http://tos-s3-cn-beijing.volces.com', s3_use_virtual_hosted_style=1, format_csv_delimiter='|', max_execution_time=18000000, receive_timeout = 10800000, send_timeout = 10800000"......clickhouse client --port 9010 --query="INSERT INTO test_elt.web_site FORMAT CSV INFILE 's3://dpsg-dataset/titan/benchmark/tpcds_csv_1000/web_site.dat' settings s3_ak_id ='', s3_ak_secret ='', s3_endpoint = 'http://tos-s3-cn-beijing.volces.com', s3_use_virtual_hosted_style=1, format_csv_delimiter='|', max_execution_time=18000000, receive_timeout = 10800000, send_timeout = 10800000"
复制代码
3.2 测试环境配置信息：
3.3 测试环境登录：
因为本人使用的是 MacOS 电脑，所以使用自带的 Shell(终端)应用进行连接到 ECS 服务器上，通过 SSH 连接到 ECS 服务器。
ssh -p 23 <提供的用户名>@<ECS服务器IP地址>
复制代码
注意，这里并不是 SSH 协议默认使用 TCP 端口 22，而是 23 端口，需要特别注意一下，在测试中为了防止使用时超时自动断开连接，这里需要使用 tmux 命令来创建一个会话。
tmux 命令普及，本人也是第一次接触这个命令，所以，顺便学习一下。
①. 因为会话的特点是窗口与其中启动的进程是连在一起的。
②. 打开窗口，会话开始；关闭窗口，会话结束，会话内部的进程也会跟着终止，不管进程有没有运行完。
③. 通过 tmux 命令会话与窗口可以 “解绑”：窗口关闭时，会话并不终止，而是继续运行，等到以后需要的时候，再让会话 “绑定” 其他窗口。
④. 允许在单个窗口中，同时访问多个会话。这对于同时运行多个命令行程序很有用。
⑤. 可以让新窗口 “接入” 已经存在的会话，还支持窗口任意的垂直和水平拆分。
⑥. 允许每个会话有多个连接窗口，因此可以多人实时共享会话。
# 创建一个新的tmux会话，$user_id是可以自定义的会话名称tmux new -s $user_id（如 tmux new -s user0001）# 后续重新登录时tmux a -t $user_id(上面这个ID)
复制代码
再执行 clickhouse client --port 9010 命令进入客户端，进入之后使用测试用数据库 test_elt，由于 TPC-DS 定义的查询语法为标准 SQL，设置数据库会话的方言类型为 ANSI。
# 使用测试用数据库 test_eltuse test_elt# 设置数据库会话的方言类型为 ANSIset dialect_type = 'ANSI'
复制代码
四、“ByConity 云原生数据仓库”MPP 架构模式与 BSP 架构模式对比：
“ByConity 云原生数据仓库”在 ELT 能力的支持，可以参考字节团队一篇文章ByConity 技术详解之 ELT，在 ByConity 内部进能够进行一定的数据处理，能够进行长任务的支持。“ByConity 云原生数据仓库”增加了 BSP 模式，可以进行 task 级别的容错，更细粒度的调度，基于资源感知的调度。
首先，我们使用命令查询一下系统相关的参数，因为参数太多了，这里主要看一下 BSP 的相关参数设置，使用以下命令即可进行查询，可以看到 bsp_mode 的 value 值是 0，表示默认是没有开启的，“If enabled, query will execute in bsp mode”。
SELECT * FROM system.settings where name like '%bsp%'
复制代码
4.1 【SQL 测试编号 1】：
从 TPC-DS 的 99 个查询中抽几个查询语句进行测试，如下为第一条语句的测试结果，我们在 2 种模式下进行了测试，可以看到结果集在 100 条的数据集进行结果的对比：
with customer_total_return as( select sr_customer_sk as ctr_customer_sk, sr_store_sk as ctr_store_sk ,sum(sr_return_amt) as ctr_total_return from store_returns, date_dim where sr_returned_date_sk = d_date_sk and d_year = 2000 group by sr_customer_sk,sr_store_sk)select c_customer_idfrom customer_total_return ctr1, store, customerwhere ctr1.ctr_total_return > ( select avg(ctr_total_return) *1.2 from customer_total_return ctr2 where ctr1.ctr_store_sk = ctr2.ctr_store_sk)and s_store_sk = ctr1.ctr_store_skand s_state = 'TN'and ctr1.ctr_customer_sk = c_customer_skorder by c_customer_idlimit 100;
复制代码
①. 可以看到在 MPP 模式下，查询的整体效果是 BSP 模式下的性能 5 倍。
②. 设置 BSP 的时候，只单纯的设置了 BSP = 1，并没有设置其它的参数。
③. 执行查询的时间，MPP 模式是 BSP 模式的 5 倍，且每秒数据处理行、每秒数据处理大小的的值也是 5 倍。
查询 work 数据的参数，我们可以看到“distributed_max_parallel_size”参数还是 0，那我们再修改一下，尝试一下。
with customer_total_return as( select sr_customer_sk as ctr_customer_sk, sr_store_sk as ctr_store_sk ,sum(sr_return_amt) as ctr_total_return from store_returns, date_dim where sr_returned_date_sk = d_date_sk and d_year = 2000 group by sr_customer_sk,sr_store_sk)select c_customer_idfrom customer_total_return ctr1, store, customerwhere ctr1.ctr_total_return > ( select avg(ctr_total_return) *1.2 from customer_total_return ctr2 where ctr1.ctr_store_sk = ctr2.ctr_store_sk)and s_store_sk = ctr1.ctr_store_skand s_state = 'TN'and ctr1.ctr_customer_sk = c_customer_skorder by c_customer_idlimit 100SETTINGS bsp_mode = 1,distributed_max_parallel_size = 12;
复制代码
将“distributed_max_parallel_size”参数设置为 12 的时候，发现执行的结果改到 26s 了，咨询官方的技术老师，相当于把一个查询切分成 N 个 task，然后分别执行，每个 task 的执行资源变少了。
4.2 【SQL 测试编号 5】：
接下来，可以使用 TPC-DS 的第 5 条 sql 语句来进行测试。
with ssr as (select s_store_id, sum(sales_price) as sales, sum(profit) as profit, sum(return_amt) as returns, sum(net_loss) as profit_loss from ( select ss_store_sk as store_sk, ss_sold_date_sk as date_sk, ss_ext_sales_price as sales_price, ss_net_profit as profit, cast(0 as decimal(7,2)) as return_amt, cast(0 as decimal(7,2)) as net_loss from store_sales union all select sr_store_sk as store_sk, sr_returned_date_sk as date_sk, cast(0 as decimal(7,2)) as sales_price, cast(0 as decimal(7,2)) as profit, sr_return_amt as return_amt, sr_net_loss as net_loss from store_returns ) salesreturns, date_dim, store where date_sk = d_date_sk and d_date between cast('2000-08-23' as date) and (cast('2000-08-23' as date) + INTERVAL '14' DAY) and store_sk = s_store_sk group by s_store_id) , csr as (select cp_catalog_page_id, sum(sales_price) as sales, sum(profit) as profit, sum(return_amt) as returns, sum(net_loss) as profit_loss from ( select cs_catalog_page_sk as page_sk, cs_sold_date_sk as date_sk, cs_ext_sales_price as sales_price, cs_net_profit as profit, cast(0 as decimal(7,2)) as return_amt, cast(0 as decimal(7,2)) as net_loss from catalog_sales union all select cr_catalog_page_sk as page_sk, cr_returned_date_sk as date_sk, cast(0 as decimal(7,2)) as sales_price, cast(0 as decimal(7,2)) as profit, cr_return_amount as return_amt, cr_net_loss as net_loss from catalog_returns ) salesreturns, date_dim, catalog_page where date_sk = d_date_sk and d_date between cast('2000-08-23' as date) and (cast('2000-08-23' as date) + INTERVAL '14' DAY) and page_sk = cp_catalog_page_sk group by cp_catalog_page_id) , wsr as (select web_site_id, sum(sales_price) as sales, sum(profit) as profit, sum(return_amt) as returns, sum(net_loss) as profit_loss from ( select ws_web_site_sk as wsr_web_site_sk, ws_sold_date_sk as date_sk, ws_ext_sales_price as sales_price, ws_net_profit as profit, cast(0 as decimal(7,2)) as return_amt, cast(0 as decimal(7,2)) as net_loss from web_sales union all select ws_web_site_sk as wsr_web_site_sk, wr_returned_date_sk as date_sk, cast(0 as decimal(7,2)) as sales_price, cast(0 as decimal(7,2)) as profit, wr_return_amt as return_amt, wr_net_loss as net_loss from web_returns left outer join web_sales on ( wr_item_sk = ws_item_sk and wr_order_number = ws_order_number) ) salesreturns, date_dim, web_site where date_sk = d_date_sk and d_date between cast('2000-08-23' as date) and (cast('2000-08-23' as date) + INTERVAL '14' DAY) and wsr_web_site_sk = web_site_sk group by web_site_id) select channel , id , sum(sales) as sales , sum(returns) as returns , sum(profit) as profit from (select 'store channel' as channel , concat('store', s_store_id) as id , sales , returns , (profit - profit_loss) as profit from ssr union all select 'catalog channel' as channel , concat('catalog_page', cp_catalog_page_id) as id , sales , returns , (profit - profit_loss) as profit from csr union all select 'web channel' as channel , concat('web_site', web_site_id) as id , sales , returns , (profit - profit_loss) as profit from wsr ) x group by rollup (channel, id) order by channel ,idlimit 100;
复制代码
在执行从 TPC-DS 的第 5 条 sql 语句时，MPP 模式下进行了报错，提示“Memory limit (total) exceeded”，内存超过最大临近值了，偶尔出现。
Received exception from server (version 21.8.7):Code: 241. DB::Exception: Received from localhost:9010. DB::Exception: Code: 241,e.displayText() = DB::Exception: Worker hos t:10.0.0.4:8124，exception:Code: 241,e.displayText() = DB::Exception: Memory limit (total) exceeded: would use 56.52 GiB (at tempt to allocate chunk of 4325376bytes)，maximum: 56.51 GiB: While executing AggregatingTransform SQLSTATE: 53000 (version21.8.7.1) SQLSTATE: 53000 (version 21.8.7.1) SQLSTATE: 53000.
复制代码
不过，我没有改参数限制查询的最大内存使用量，如下为第 5 条语句的测试结果，我们在 2 种模式下进行了测试，可以看到结果集在 100 条的数据集进行结果的对比：
①. 可以看到，现在的结果是相反的，在 BSP 模式下，查询的整体效果是 MPP 模式下的性能 7 倍。
②. 设置 BSP 的时候，设置了 BSP = 1，将“distributed_max_parallel_size”参数设置为 12 的。
③. 执行查询的时间，BSP 模式是 MPP 模式的 7 倍，且每秒数据处理行、每秒数据处理大小的的值也是 7 倍。
对于大的数据量，BSP 通过把一个查询切换成 N 个 task，可以加快查询的效率。
4.3 【SQL 测试编号 11】：
接下来，可以使用 TPC-DS 的第 11 条 sql 语句来进行测试。
with year_total as ( select c_customer_id customer_id ,c_first_name customer_first_name ,c_last_name customer_last_name ,c_preferred_cust_flag customer_preferred_cust_flag ,c_birth_country customer_birth_country ,c_login customer_login ,c_email_address customer_email_address ,d_year dyear ,sum(ss_ext_list_price-ss_ext_discount_amt) year_total ,'s' sale_type from customer ,store_sales ,date_dim where c_customer_sk = ss_customer_sk and ss_sold_date_sk = d_date_sk group by c_customer_id ,c_first_name ,c_last_name ,c_preferred_cust_flag ,c_birth_country ,c_login ,c_email_address ,d_year union all select c_customer_id customer_id ,c_first_name customer_first_name ,c_last_name customer_last_name ,c_preferred_cust_flag customer_preferred_cust_flag ,c_birth_country customer_birth_country ,c_login customer_login ,c_email_address customer_email_address ,d_year dyear ,sum(ws_ext_list_price-ws_ext_discount_amt) year_total ,'w' sale_type from customer ,web_sales ,date_dim where c_customer_sk = ws_bill_customer_sk and ws_sold_date_sk = d_date_sk group by c_customer_id ,c_first_name ,c_last_name ,c_preferred_cust_flag ,c_birth_country ,c_login ,c_email_address ,d_year ) select t_s_secyear.customer_id ,t_s_secyear.customer_first_name ,t_s_secyear.customer_last_name ,t_s_secyear.customer_preferred_cust_flag from year_total t_s_firstyear ,year_total t_s_secyear ,year_total t_w_firstyear ,year_total t_w_secyear where t_s_secyear.customer_id = t_s_firstyear.customer_id and t_s_firstyear.customer_id = t_w_secyear.customer_id and t_s_firstyear.customer_id = t_w_firstyear.customer_id and t_s_firstyear.sale_type = 's' and t_w_firstyear.sale_type = 'w' and t_s_secyear.sale_type = 's' and t_w_secyear.sale_type = 'w' and t_s_firstyear.dyear = 2001 and t_s_secyear.dyear = 2001+1 and t_w_firstyear.dyear = 2001 and t_w_secyear.dyear = 2001+1 and t_s_firstyear.year_total > 0 and t_w_firstyear.year_total > 0 and case when t_w_firstyear.year_total > 0 then t_w_secyear.year_total / t_w_firstyear.year_total else 0.0 end > case when t_s_firstyear.year_total > 0 then t_s_secyear.year_total / t_s_firstyear.year_total else 0.0 end order by t_s_secyear.customer_id ,t_s_secyear.customer_first_name ,t_s_secyear.customer_last_name ,t_s_secyear.customer_preferred_cust_flaglimit 100;
复制代码
在执行从 TPC-DS 的第 11 条 sql 语句时，在 MPP 模式下，第一次执行的时间是 34s，上面可以通过使用 BSP 模式来进行增加 task 任务的消费，那么我们这里只是单纯的增加内存，会不会有所改善？可以看到结果集在 100 条的数据集进行结果的对比：
通过 4 种模式的同一个 SQL11 语句测试对比，接下来比较一下 MPP 模式和 BSP 模式各种参数的执行结果对比：
①. 可以看到，相对于上面对比 SQL5 处理的 7.2 亿的数据，而 SQL11 处理的是 0.12 亿的数据，这次 BSP 模式显示处理的优势弱于 MPP 模式。
②. 设置 BSP 的时候，我们将“distributed_max_parallel_size”参数设置为默认值、12、24 来进行阶段性测试，可以看到在有限的数据集中(0.12 亿)，并不是切分的 task 越多，处理数据的速度越快。
③. 执行查询的时间，BSP 模式 worker 12 个跟 worker24 的趋势比例也是差不多接近 2 倍。
4.4 程序监控趋势图：
因为看不到 Prometheus 类似监控系统，有一些指标参数不太好进行参考，所以，自己写了一个对 clickhouse 这个进程的监控，收集一下 CPU 和内存的的使用情况，以便于对整个 ByConity 的性能方面的分析，以下为 shell 脚本内容：
#!/bin/bash
# 无限循环，每秒记录一次while true; do # 获取进程的CPU和内存使用情况 ps aux --no-headers | awk '{print $2, $3, $4, $11}' | grep clickhouse >> cpu_mem_usage.log
# 等待一秒 sleep 1done
复制代码
①. 可以看到，在整个过程中，绿色的折现图是表示 CPU 的实时走势图，而蓝色是表示内存的实时走势图。
②. 在 CPU 的实时走势图，可以看到整个过程中，Max 是可以达到 10%左右，基本上还是比较稳定在 2% - 4%左右。
③. 在内存的实时走势图，可以看到整个过程中，Max 是可以达到 1.5 左右，基本上还是比较稳定。
当然，这里没有实时的去查询，就是本人自己在控制台进行一下测试，希望下次写个脚本去进行循环测试，当然，更详细的监控记录更为方便。
4.5 ByConity 使用 ELT 相关参数：
在 query settings 中通过以下参数使用 ELT 能力：
五、BSP 技术总结：
5.1 业务痛点：
公司从最早期的 excel 统计时代，到后面成立技术部，经过公司不同的发展阶段，也迭代开发出了几个版本的数字化平台体系建设，在目前现有的 OLTP 数据分析场景中，将数据统一汇聚到一个数据 BI 系统平台中。
在此过程中，企业的数据分散于多个系统中，由于数据标准不统一，时间标签缺失等问题，导致数据质量问题。另外，随着企业的业务技术改造和业务流程的不断变化，原有数据可能失去时效性而不再适用，容易造成“假数据”、“脏数据”、“死数据”。
针对复杂查询，尤其是数据仓库中典型的 ETL 任务来说，ClickHouse 则并不擅长，结构复杂、耗时较长的数据加工作业，通常需要复杂的调优过程。典型的问题如下：
①. 重试成本高:对于运行时长在分钟级甚至小时级的 ETL 作业，如果运行过程中出现失败，ClickHouse 只能进行 query 级别的重试。从头重试不仅造成大量的资源浪费，也对加工任务的 SLA 提出了挑战。
②. 资源占用巨大:由于缺少迭代计算和有效的任务拆分，在查询数据量大、计算复杂的情况下，通常要求节点有充足的内存进行处理。
③. 并发控制:当多个查询同时运行时，ClickHouse 并不会根据资源的使用情况进行调度。任务之间相互挤压会导致失败(通常是 Memorylimit 错误)。叠加重试机制的缺乏，通常会引起雪崩效应。
2. ByConity 增加 BSP 模式：
ByConity 在 ClickHouse 高性能计算框架的基础上，增加了对 bsp 模式的支持，可以进行 task 级别的容错;更细粒度的调度，支持资源感知的调度：
①. 当 query 运行中遇到错误时，可以自动重试当前的 task，而不是从头进行重试。大大减少重试成本。
②. 当 query 需要的内存巨大，甚至大于单机的内存时，可以通过增加并行度来减少单位时间内内存的占用。只需要调大并行度参数即可，理论上是可以无限扩展的。
③. 可以根据集群资源使用情况有序调度并发 ETL 任务，从而减少资源的挤占，避免频繁失败。
④. 对于大的数据量，BSP 通过把一个查询切换成 N 个 task，可以加快查询的效率。
⑤. 通过 BSP 模式的效果非常的明显，通过 BSP 能力，把数据加工（T）的过程转移到 ByConity 内部，能够一站式完成数据接入、加工和分析。
⑥. 可以在 ByConity 开启几个不同的 vw，比如：vw1 专门跑批量数据，vw2 实时查询，vw2 专门使用 mpp 模式，通过这样拆分任务的方式，将需要查询最快时间查出来。然后 vw1 把表 ODS 数据清洗，加工到 vw2 的 DWD 里，然后 vw2 直接查 DWD 的数据。
评论