随着大数据技术的发展，实时数据仓库和离线数据仓库在企业数据分析中的重要性日益增加。为了满足企业对数据处理性能和效率的多样化需求，ByConity 作为一款开源云原生数据仓库，提供了一个高效的解决方案，特别是在 ELT（Extract, Load, Transform） 任务的执行上。与传统的 ETL（Extract, Transform, Load）模式不同，ByConity 通过优化的架构和灵活的处理方式，能够显著提升数据处理的效率与稳定性。

本文将通过 ByConity 在 ELT 测试中的应用案例，深入探讨其架构特性、性能优化和实战经验，并为读者展示如何通过 ByConity 提高数据仓库的性能和灵活性。

ByConity 架构与 ELT 模式的优势

ByConity 采用了云原生架构，采用了存算分离的设计理念，能够为用户提供灵活的扩展能力和高效的资源利用。ByConity 的 ELT 实现可以简化数据处理复杂性，提升实时和批量数据处理能力。

1. 数据提取（Extract） ByConity 支持从各种数据源（如关系型数据库、文件系统、消息队列等）提取数据。其强大的连接器和数据读取接口，能够高效地从异构系统中提取大量数据。

2. 数据加载（Load） 通过支持批量加载和实时加载的方式，ByConity 可以快速地将数据写入数据仓库。其高效的数据加载机制能够保证数据的准确性和一致性，避免加载过程中的性能瓶颈。

3. 数据转换（Transform） 在 ByConity 中，数据转换操作可以通过 SQL 查询或其他数据处理语言完成。它提供了强大的计算引擎和丰富的函数库，支持数据清洗、数据聚合等复杂操作。这一过程能够充分利用目标系统的计算资源，避免传统 ETL 过程中复杂的转换任务。

核心优势

ByConity 是分布式的云原生 SQL 数仓引擎，擅长交互式查询和即席查询，具有支持多表关联复杂查询、集群扩容无感、离线批数据和实时数据流统一汇总等特点。

高性能低成本

通过向量化执行引擎、列式存储和 CBO+RBO 优化器的结合，系统能够在海量数据规模下实现亚秒级查询响应能力。这使得在大数据量处理时，依然能够维持高效的查询性能。与此同时，系统提供超高压缩比率，大大节省了存储空间，降低了磁盘成本。这种存储压缩不仅提高了数据处理的效率，还显著减少了数据存储的开销。

多种场景统一支持

系统支持实时数据流和离线批数据写入，能够灵活处理不同数据类型的输入需求。它不仅具备交互式事务能力，能够处理多表关联查询，还能满足在线系统中对交互式查询的高要求。同时，它也适用于后台的实时监控、报表大屏等场景。无论是快速响应的实时查询，还是复杂的批量数据处理，系统都能够提供优异的支持，确保数据处理的高效性和准确性。

生态友好

该系统兼容 ClickHouse 大多数接口和工具，提供对 Kafka、Spark、Flink 等多种数据导入的支持，能够无缝集成到现有的数据处理生态中。对于数据可视化工具，也有广泛的兼容性，支持 Superset、Tableau 等流行的可视化工具，方便用户进行数据分析和展示。这种生态友好的设计，使得该系统能够轻松融入到各种现有的大数据架构和数据流中，降低了使用门槛并提高了数据处理的灵活性和效率。

ByConity 功能特性概述

1. 弹性扩缩容 ByConity 采用存储计算分离的架构，适合动态扩缩容需求的场景。通过分离的元数据和数据存储，计算节点无状态化，使扩缩容变得轻量，计算实例启动后即刻生效，无需额外的数据迁移开销，从而实现实时扩缩容。

2. 多租户隔离和资源共享 ByConity 支持为每个查询 SQL 指定计算组，从而实现物理资源隔离，避免不同租户间的查询干扰。同时，系统支持计算组之间的资源租借，提升资源利用率。

   
   

3. 读写分离 通过存储计算分离架构，ByConity 原生支持读写分离。Insert 操作使用专门的写入计算组，而 Select 查询则使用专门的读取计算组，避免读写作业相互影响，优化系统性能。

4. 查询优化器

   CBO（基于成本的优化）：通过收集数据库统计信息，评估不同执行计划的成本，并选择成本最低的计划。支持多种优化技术，如 Join/Agg Reorder、CTE、动态过滤推送等。

   RBO（基于规则的优化）：通过启发式规则进行优化，包括列裁剪、子查询解除、冗余运算符消除等。

   DBO（基于数据依赖的优化）：基于数据依赖关系进行优化，支持唯一键、函数依赖等。

5. 查询调度 ByConity 提供两种查询调度策略：

   Cache-aware 调度：针对存储计算分离场景，最大化 Cache 利用，减少冷读，提升性能。

   Resource-aware 调度：感知集群资源使用情况，进行高效调度，确保资源合理分配并进行流量控制，避免系统过载。

6. 数据湖支持 在 0.2.0 版本中，ByConity 支持通过外表访问 Hive 数据，支持的存储系统包括 HDFS 和 S3，文件格式支持 Parquet 和 ORC，并支持磁盘缓存（Disk Cache）。此外，还支持 Hudi 外表，包括 copy-on-write 和 merge-on-write 两种模式，适用于数据湖的高效操作。

   
   

7. ELT 支持 ByConity 在 0.2.0 版本中支持了部分 ELT 能力，如异步能力、执行队列和基于硬盘的 shuffle。0.3.0 版本引入了新的 BSP（Bulk Synchronous Parallel）模式，优化了硬盘洗牌的效率，提高了吞吐量。

   
   

8. 倒排索引 从 0.3.0 版本起，ByConity 支持倒排索引，显著增强了其在文本检索领域的能力，尤其在进行日志数据分析等需要高性能查询的场景中，能够提供显著提升。

这些特性表明 ByConity 具备高效的扩展性、灵活的资源管理和强大的数据处理能力，适合多种场景下的企业级应用，尤其是数据密集型和多租户环境。

登录 ECS 服务器

1. MacOS / Linux 用户

在 MacOS 或 Linux 系统上，通过终端（Shell）应用程序使用 SSH 连接。按照以下步骤操作：

1. 打开终端 打开终端应用程序。

2. 输入 SSH 命令连接 在终端中输入以下命令连接，并将 <提供的用户名> 和 <服务器IP地址> 替换为实际的用户名和服务器 IP 地址：

   ssh -p 23 <提供的用户名>@<IP 地址>

   按回车键。

3. 确认连接 系统可能会提示确认是否继续连接，输入 yes 并按回车键确认。

4. 输入登录密码 系统会提示输入密码，输入提供的密码并按回车键。

5. 避免会话超时 为了避免会话因超时断开，使用 tmux 创建一个新的会话。输入以下命令来启动一个新的 tmux 会话：

   tmux new -s $user_id

   其中，$user_id 是自定义的会话名称（例如 user0001）。

   如果需要恢复之前的会话，可以使用以下命令：

   tmux a -t $user_id

6. 进入 ClickHouse 客户端 在创建会话并登录后，输入以下命令进入 ClickHouse 客户端：

   clickhouse client --port 9010

   如果后续的 SQL 输入被截断，可以使用如下命令来避免截断：

   clickhouse client --port 9010 -mn

   SQL 查询需要以分号 ; 结束。

2. Windows 用户

Windows 10 和 Windows 11 系统内置了 OpenSSH 客户端，可以通过命令提示符（CMD）连接到 ECS 服务器。

按照以下步骤操作：

1. 打开命令提示符 打开 Windows “开始菜单”，搜索并打开“命令提示符”应用。

2. 输入 SSH 命令连接到服务器 在命令提示符中，输入以下命令连接到 ECS 服务器，将 <提供的用户名> 和 <ECS服务器IP地址> 替换为实际的用户名和服务器 IP 地址：

   ssh -p 23 <提供的用户名>@<ECS 服务器 IP 地址>

   按回车键。

3. 确认连接 系统将提示确认是否连接到该服务器，输入 yes 并按回车键确认。

4. 输入登录密码 系统会提示输入密码，输入提供的密码并按回车键。

5. 避免会话超时 为了避免会话超时导致断开，建议使用 tmux 创建一个新的会话。输入以下命令：

   tmux new -s $user_id

   其中，$user_id 是自定义的会话名称，例如 user0001。

   如果需要恢复之前的会话，请使用以下命令：

   tmux a -t $user_id

6. 进入 ClickHouse 客户端 在创建会话并登录后，输入以下命令进入 ClickHouse 客户端：

   clickhouse client --port 9010

   如果 SQL 输入被截断，可以使用如下命令避免截断：

   clickhouse client --port 9010 -mn

   SQL 查询需要以分号 ; 结束。

TPC-DS 1TB 数据集 的测试活动

测试环境

首先需要 ssh 链接。

为了避免在使用过程中因超时而自动断开连接，可以通过运行以下命令创建一个新的会话：

tmux new -s $user_id

例如：

tmux new -s user001

其中，$user_id 可以自定义为任何喜欢的会话名称。

接下来，我们可以通过运行以下命令进入客户端：

clickhouse client --port 9010

该命令将启动 ClickHouse 客户端并连接到指定的端口。

使用测试用数据库 test_elt，可以执行以下命令：

use test_elt;

此命令会将当前数据库切换到 test_elt，之后就可以在该数据库中执行查询操作。

为了确保使用 TPC-DS 定义的查询语法符合标准 SQL，我们需要将数据库会话的方言类型设置为 ANSI。可以执行以下命令：

set dialect_type = 'ANSI';

此命令将会话的方言类型设置为 ANSI，从而确保后续的查询符合标准 SQL 语法。

执行以下 SQL 查询时，由于内存限制，可能会导致执行失败。

with ws as        (select d_year AS ws_sold_year, ws_item_sk,        ws_bill_customer_sk ws_customer_sk,        sum(ws_quantity) ws_qty,        sum(ws_wholesale_cost) ws_wc,        sum(ws_sales_price) ws_sp        from web_sales        left join web_returns on wr_order_number=ws_order_number and ws_item_sk=wr_item_sk        join date_dim on ws_sold_date_sk = d_date_sk        where wr_order_number is null        group by d_year, ws_item_sk, ws_bill_customer_sk        ),        cs as        (select d_year AS cs_sold_year, cs_item_sk,        cs_bill_customer_sk cs_customer_sk,        sum(cs_quantity) cs_qty,        sum(cs_wholesale_cost) cs_wc,        sum(cs_sales_price) cs_sp        from catalog_sales        left join catalog_returns on cr_order_number=cs_order_number and cs_item_sk=cr_item_sk        join date_dim on cs_sold_date_sk = d_date_sk        where cr_order_number is null        group by d_year, cs_item_sk, cs_bill_customer_sk        ),        ss as        (select d_year AS ss_sold_year, ss_item_sk,        ss_customer_sk,        sum(ss_quantity) ss_qty,        sum(ss_wholesale_cost) ss_wc,        sum(ss_sales_price) ss_sp        from store_sales        left join store_returns on sr_ticket_number=ss_ticket_number and ss_item_sk=sr_item_sk        join date_dim on ss_sold_date_sk = d_date_sk        where sr_ticket_number is null        group by d_year, ss_item_sk, ss_customer_sk        )        select        ss_sold_year, ss_item_sk, ss_customer_sk,        round(ss_qty/(coalesce(ws_qty,0)+coalesce(cs_qty,0)),2) ratio,        ss_qty store_qty, ss_wc store_wholesale_cost, ss_sp store_sales_price,        coalesce(ws_qty,0)+coalesce(cs_qty,0) other_chan_qty,        coalesce(ws_wc,0)+coalesce(cs_wc,0) other_chan_wholesale_cost,        coalesce(ws_sp,0)+coalesce(cs_sp,0) other_chan_sales_price        from ss        left join ws on (ws_sold_year=ss_sold_year and ws_item_sk=ss_item_sk and ws_customer_sk=ss_customer_sk)        left join cs on (cs_sold_year=ss_sold_year and cs_item_sk=ss_item_sk and cs_customer_sk=ss_customer_sk)        where (coalesce(ws_qty,0)>0 or coalesce(cs_qty, 0)>0) and ss_sold_year=2000        order by        ss_sold_year, ss_item_sk, ss_customer_sk,        ss_qty desc, ss_wc desc, ss_sp desc,        other_chan_qty,        other_chan_wholesale_cost,        other_chan_sales_price,        ratio        LIMIT 100        SETTINGS bsp_mode = 1,distributed_max_parallel_size = 12;

该 SQL 查询的目的是比较 2000 年在不同销售渠道（web_sales、catalog_sales、`store_sales）中的销售数据，并计算每个商品在店内销售与其他渠道（网站和目录）销售之间的比例，最终筛选出销售数量较高的记录。

主要步骤：

1. 定义三个临时表：

   ws（Web Sales）：计算 2000 年每个商品在web_sales中的销售数据，排除已退货的订单（web_returns）。计算销售数量（ws_qty）、批发成本（ws_wc）和销售金额（ws_sp），并按商品和客户分组。

   cs（Catalog Sales）：计算 2000 年每个商品在catalog_sales中的销售数据，排除已退货的订单（catalog_returns）。同样，计算销售数量（cs_qty）、批发成本（cs_wc）和销售金额（cs_sp），并按商品和客户分组。

   ss（Store Sales）：计算 2000 年每个商品在store_sales中的销售数据，排除已退货的订单（store_returns）。计算销售数量（ss_qty）、批发成本（ss_wc）和销售金额（ss_sp），并按商品和客户分组。

2. 主查询：

   将三个临时表 (ws、cs 和 ss) 进行左连接，基于年份（d_year）、商品 ID（item_sk）和客户 ID（customer_sk）进行匹配。

   计算每个商品在商店销售中的数量与其他渠道销售（

   web_sales

   和

   catalog_sales

   ）数量的比例：

   ratio = ss_qty / (coalesce(ws_qty, 0) + coalesce(cs_qty, 0))：商店销售数量与其他渠道销售数量的比率。

   计算商店销售数量、批发成本和销售金额，同时计算其他渠道的销售数量、批发成本和销售金额（coalesce函数用于处理空值）。

3. 筛选和排序：

   仅保留 2000 年的记录，且在其他渠道（Web 和 Catalog）有销售数据的商品（即coalesce(ws_qty, 0) > 0 或 coalesce(cs_qty, 0) > 0）。

   按照销售数量（ss_qty）、批发成本（ss_wc）、销售金额（ss_sp）、其他渠道销售数量、批发成本和销售金额以及比例（ratio）进行排序，优先显示销售数量较高的商品。

4. 限制结果：

   查询返回前 100 条记录，限制结果集。

5. 设置查询参数：

   SETTINGS bsp_mode = 1, distributed_max_parallel_size = 12：设置查询的并行执行和分布式查询参数，优化查询性能。

该查询旨在比较 2000 年商店销售和其他渠道（Web、目录销售）的数据，计算商店销售数量与其他渠道销售数量的比例，筛选出在其他渠道也有销售的商品，并按销售数量、销售金额等进行排序，最终返回前 100 条销售数据。

由于内存限制，查询无法执行失败。在 SQL 查询失败后，可以在查询末尾添加以下设置并重新执行：

（注意：在 SQL 拼接时，末尾的 ; 需要去掉）

SETTINGS bsp_mode = 1, distributed_max_parallel_size = 4;

我们在此将 distributed_max_parallel_size 设置为 4（请确保设置为 4 的倍数），如果仍然无法解决问题，可能需要进一步调整。

在执行以下查询时，可以通过添加参数来限制查询的最大内存使用量。例如：

SETTINGS max_memory_usage = 40000000000;

（单位为字节，当前值约为 37.25 GB）

通过设置合适的内存限制，可以防止内存溢出（OOM）。如果内存设置过高，仍然可能会引发内存溢出错误，需要进一步调整内存限制。

WITH all_sales AS (        SELECT d_year        ,i_brand_id        ,i_class_id        ,i_category_id        ,i_manufact_id        ,SUM(sales_cnt) AS sales_cnt        ,SUM(sales_amt) AS sales_amt        FROM (SELECT d_year        ,i_brand_id        ,i_class_id        ,i_category_id        ,i_manufact_id        ,cs_quantity - COALESCE(cr_return_quantity,0) AS sales_cnt        ,cs_ext_sales_price - COALESCE(cr_return_amount,0.0) AS sales_amt        FROM catalog_sales JOIN item ON i_item_sk=cs_item_sk        JOIN date_dim ON d_date_sk=cs_sold_date_sk        LEFT JOIN catalog_returns ON (cs_order_number=cr_order_number        AND cs_item_sk=cr_item_sk)        WHERE i_category='Books'        UNION        SELECT d_year        ,i_brand_id        ,i_class_id        ,i_category_id        ,i_manufact_id        ,ss_quantity - COALESCE(sr_return_quantity,0) AS sales_cnt        ,ss_ext_sales_price - COALESCE(sr_return_amt,0.0) AS sales_amt        FROM store_sales JOIN item ON i_item_sk=ss_item_sk        JOIN date_dim ON d_date_sk=ss_sold_date_sk        LEFT JOIN store_returns ON (ss_ticket_number=sr_ticket_number        AND ss_item_sk=sr_item_sk)        WHERE i_category='Books'        UNION        SELECT d_year        ,i_brand_id        ,i_class_id        ,i_category_id        ,i_manufact_id        ,ws_quantity - COALESCE(wr_return_quantity,0) AS sales_cnt        ,ws_ext_sales_price - COALESCE(wr_return_amt,0.0) AS sales_amt        FROM web_sales JOIN item ON i_item_sk=ws_item_sk        JOIN date_dim ON d_date_sk=ws_sold_date_sk        LEFT JOIN web_returns ON (ws_order_number=wr_order_number        AND ws_item_sk=wr_item_sk)        WHERE i_category='Books') sales_detail        GROUP BY d_year, i_brand_id, i_class_id, i_category_id, i_manufact_id)        SELECT prev_yr.d_year AS prev_year        ,curr_yr.d_year AS year        ,curr_yr.i_brand_id        ,curr_yr.i_class_id        ,curr_yr.i_category_id        ,curr_yr.i_manufact_id        ,prev_yr.sales_cnt AS prev_yr_cnt        ,curr_yr.sales_cnt AS curr_yr_cnt        ,curr_yr.sales_cnt-prev_yr.sales_cnt AS sales_cnt_diff        ,curr_yr.sales_amt-prev_yr.sales_amt AS sales_amt_diff        FROM all_sales curr_yr, all_sales prev_yr        WHERE curr_yr.i_brand_id=prev_yr.i_brand_id        AND curr_yr.i_class_id=prev_yr.i_class_id        AND curr_yr.i_category_id=prev_yr.i_category_id        AND curr_yr.i_manufact_id=prev_yr.i_manufact_id        AND curr_yr.d_year=2002        AND prev_yr.d_year=2002-1        AND CAST(curr_yr.sales_cnt AS DECIMAL(17,2))/CAST(prev_yr.sales_cnt AS DECIMAL(17,2))<0.9        ORDER BY sales_cnt_diff,sales_amt_diff        limit 100        SETTINGS max_memory_usage=20000000000        SETTINGS bsp_mode = 1,distributed_max_parallel_size = 12;

该 SQL 查询汇总了 2002 年和 2001 年图书类商品在不同销售渠道（目录、商店、网络）的销售数据，并计算了每个品牌、类别、制造商和分类的销售数量和金额差异。通过比较当前年份（2002 年）与前一年（2001 年）的销售数据，筛选出销售数量下降超过 10%的商品，并按销售数量和金额的差异进行排序，最终返回销售下降幅度较大的 100 条记录。查询通过设置内存限制和并行执行参数，以优化性能。

SQL 查询执行与故障排除

在测试过程中，我执行了多个复杂的 SQL 查询，以下是针对每个查询的详细分析和故障排除过程：

第一个查询：比较不同销售渠道的销售数据

查询的目标是比较 2000 年不同销售渠道（Web、Catalog、Store）的销售数据，并计算每个商品在店内销售与其他渠道销售之间的比例。该查询的执行步骤包括：

• 临时表定义： 通过 ws（Web Sales）、cs（Catalog Sales）和 ss（Store Sales）三个临时表分别计算各个渠道的销售数量、批发成本和销售价格。

• 左连接操作： 将三个临时表（ws、cs 和 ss）根据年份、商品 ID 和客户 ID 进行左连接，以便计算每个商品在商店销售和其他渠道销售之间的比例。

• 结果筛选与排序： 结果按销售数量、批发成本、销售金额等进行排序，最终筛选出销售数量较高的前 100 条记录。

• 查询设置： 使用了以下 SQL 设置以优化查询性能：

  SETTINGS bsp_mode = 1, distributed_max_parallel_size = 12;

  这两个设置分别控制了并行执行模式和最大并行任务数。由于数据量巨大，查询执行可能遇到内存不足的问题。

内存问题与解决

执行查询时，由于数据集庞大，可能会遇到内存不足（OOM）的问题。针对这种情况，使用了以下两种方法进行优化：

1. 减少并行执行的任务数：

   SETTINGS bsp_mode = 1, distributed_max_parallel_size = 4;

   将 distributed_max_parallel_size 设置为 4，可以降低每个节点处理的并行任务数，从而减少内存占用。

2. 限制查询的最大内存使用量： 为了避免内存溢出，设置了查询的最大内存使用量：

   SETTINGS max_memory_usage = 40000000000;

   该设置将查询的内存上限限制为 40 GB。如果内存使用超过此限制，查询会提前终止，避免系统崩溃。

第二个查询：比较前后两年销售数据

在第二个查询中，聚合了 2002 年和 2001 年图书类商品在不同销售渠道的销售数据，并计算了销售数量和金额的差异。具体步骤如下：

• 子查询（UNION 操作）： 通过联合多个销售数据表（catalog_sales、store_sales 和 web_sales），计算了不同销售渠道中的销售数量（sales_cnt）和销售金额（sales_amt），并去除了已退货的数据。

• 计算前后两年差异： 比较 2002 年和 2001 年的销售数据，筛选出那些销售数量下降超过 10%（sales_cnt_diff < 0）的商品。查询返回的是销售数量下降最大的 100 条记录。

• 性能优化： 通过以下设置优化了查询性能：

  SETTINGS max_memory_usage=20000000000; SETTINGS bsp_mode = 1, distributed_max_parallel_size = 12;

  同样，设置了内存限制为 20 GB，确保查询不会因内存不足而崩溃。同时，通过增加并行执行任务的数量，进一步加速查询。

其他性能优化建议

为了确保在处理大规模数据集时，SQL 查询能够高效执行，还可以采取以下优化措施：

• 调整数据分区： 对于极大的数据集，可以考虑将数据进行合理的分区。通过按日期、商品类别等字段进行分区，减少查询时的数据扫描量，优化查询效率。

• 使用物化视图： 对于频繁使用的查询，可以考虑使用物化视图（Materialized Views）。通过提前计算并存储查询结果，减少查询时的计算负担。

• 索引优化： 确保在查询中涉及的字段（如 item_sk、customer_sk、d_year 等）上建立合适的索引，以提高查询速度。虽然 ClickHouse 是列式数据库，但合理的索引配置依然能够显著提高查询性能。

心得

此次测试活动验证了 ByConity 集群在处理 TPC-DS 1TB 数据集时的性能表现。在执行复杂查询时，内存限制和并行计算设置是优化查询性能的关键。通过合适的 SQL 设置、内存管理和资源调度，能够确保在大数据量下依然保持较高的查询效率。

在实际操作中，可以通过以下方式进一步提升性能：

• 调整并行执行任务数：根据服务器资源和查询复杂度灵活调整 distributed_max_parallel_size。

• 使用内存限制：根据查询数据量设置合理的 max_memory_usage，避免内存溢出。

• 优化查询计划：针对常见查询，使用物化视图或缓存查询结果，减少重复计算。

最后，测试过程中遇到的内存问题和查询失败可以通过适当调整 SQL 设置来解决。确保在执行较大数据集查询时，时刻关注内存使用情况和集群资源。

总结与展望

通过 ByConity 的 ELT 测试，我们验证了其在处理大规模数据时的高效性和稳定性。特别是在实时数据仓库和离线数仓的应用场景中，ByConity 凭借其分布式架构和优化的计算资源调度，能够有效提升数据处理的效率。

ByConity 的开源特性使得其在数据分析平台中具有广泛的应用前景，无论是在线实时分析还是离线数据加工，均能提供强大的支持。未来，随着集群资源管理和查询优化技术的进一步提升，ByConity 将在数据仓库领域发挥更大的潜力，帮助企业实现更高效、更可靠的数据分析和决策支持。

附录

ByConity 技术文档：什么是ByConity | ByConity

ByConity：ByConity | ByConity