1. 背景
火山引擎增長分析 DataFinder 基于 ClickHouse 來進行行為日志的分析，ClickHouse 的主要版本是基于社區版改進開發的字節內部版本。主要的表結構：
事件表：存儲用戶行為數據，以用戶 ID 分 shard 存儲。
--列出了主要的字段信息
CREATE TABLE tob_apps_all
(
`tea_app_id` UInt32, --應用ID
`device_id` String DEFAULT '', --設備ID
`time` UInt64,--事件日志接受時間
`event` String,--事件名稱
`user_unique_id` String,--用戶ID
`event_date` Date , --事件日志日期，由time轉換而來
`hash_uid` UInt64 --用戶ID hash過后的id，用來join降低內存消耗
)│
```
用戶表：存儲用戶的屬性數據，以用戶 ID 分 shard 存儲。
--列出了主要的字段信息
CREATE TABLE users_unique_all
(
`tea_app_id` UInt32, --應用ID
`user_unique_id` String DEFAULT '', -- 用戶ID
`device_id` String DEFAULT '', -- 用戶最近的設備ID
`hash_uid` UInt64,--用戶ID hash過后的id，用來join降低內存消耗
`update_time` UInt64,--最近一次更新時間
`last_active_date` Date --用戶最后活躍日期
)
設備表：存儲設備相關的數據，以設備 ID 分 shard 存儲。
--列出了主要的字段信息
CREATE TABLE devices_all
(
`tea_app_id` UInt32, --應用ID
`device_id` String DEFAULT '', --設備ID
`update_time` UInt64, --最近一次更新時間
`last_active_date` Date --用戶最后活躍日期
)
業務對象表：存儲業務對象相關的數據，每個 shard 存儲全量的數據。
--列出了主要的字段信息
CREATE TABLE rangers.items_all
(
`tea_app_id` UInt32,
`hash_item_id` Int64,
`item_name` String, --業務對象名稱。比如商品
`item_id` String, --業務對象ID。比如商品id 1000001
`last_active_date` Date
)
1.1 業務挑戰
隨著接入應用以及應用的 DAU 日益增加，ClickHouse 表的事件量增長迅速；并且基于行為數據需要分析的業務指標越來越復雜，需要 JOIN 的表增多；我們遇到有一些涉及到 JOIN 的復雜 SQL 執行效率低，內存和 CPU 資源占用高，導致分析接口響應時延和錯誤率增加。
2. 關于 Clickhouse 的 JOIN
在介紹優化之前，先介紹一下基本的 ClickHouse JOIN 的類型和實現方式。
2.1 分布式 JOIN
SELECT
et.os_name,
ut.device_id AS user_device_id
FROM tob_apps_all AS et
ANY LEFT JOIN
(
SELECT
device_id,
hash_uid
FROM users_unique_all
WHERE (tea_app_id = 268411) AND (last_active_date >= '2022-08-06')
) AS ut ON et.hash_uid = ut.hash_uid
WHERE (tea_app_id = 268411)
AND (event = 'app_launch')
AND (event_date = '2022-08-06')
基本執行過程：
一個 Clickhouse 節點作為 Coordinator 節點，給每個節點分發子查詢，子查詢 sql（tob_apps_all 替換成本地表，users_unique_all 保持不變依然是分布式表）。
每個節點執行 Coordinator 分發的 sql 時，發現 users_unique_all 是分布式表，就會去所有節點上去查詢以下 SQL(一共有 N*N。N 為 shard 數量)。
- SELECT device_id, hash_uid FROMusers_uniqueWHERE (tea_app_id = 268411) AND (last_active_date >= '2022-08-06')
- 每個節點從其他 N-1個節點拉取2中子查詢的全部數據，全量存儲(內存 or 文件)，進行本地 JOIN。
- Coordinator 節點從每個節點拉取3中的結果集，然后做處理返回給 client。
存在的問題：
- 子查詢數量放大。
- 每個節點都全量存儲全量的數據。
2.2 分布式 Global JOIN
SELECT
et.os_name,
ut.device_id AS user_device_id
FROM tob_apps_all AS et
GLOBAL ANY LEFT JOIN
(
SELECT
device_id,
hash_uid
FROM users_unique_all
WHERE (tea_app_id = 268411) AND (last_active_date >= '2022-08-06')
) AS ut ON et.hash_uid = ut.hash_uid
WHERE (tea_app_id = 268411)
AND (event = 'app_launch')
AND (event_date = '2022-08-06')
基本執行過程：
- 一個 Clickhouse 節點作為 Coordinator 節點，分發查詢。在每個節點上執行sql（tob_apps_all 替換成本地表，右表子查詢替換成別名 ut）。
- Coordinator 節點去其他節點拉取 users_unique_all 的全部數據，然后分發到全部節點（作為1中別名表 ut 的數據）。
- 每個節點都會存儲全量的2中分發的數據(內存or文件)，進行本地 local join。
- Coordinator 節點從每個節點拉取3中的結果集，然后做處理返回給 client。
存在的問題：
- 每個節點都全量存儲數據。
- 如果右表較大，分發的數據較大，會占用網絡帶寬資源。
2.3 本地 JOIN
SQL 里面只有本地表的 JOIN，只會在當前節點執行。
SELECT et.os_name,ut.device_id AS user_device_id
FROM tob_apps et any LEFT JOIN
(SELECT device_id,
hash_uid
FROM rangers.users_unique
WHERE tea_app_id = 268411
AND last_active_date>='2022-08-06') ut
ON et.hash_uid=ut.hash_uid
WHERE tea_app_id = 268411
AND event='app_launch'
AND event_date='2022-08-06'
2.3.1 Hash join
- 右表全部數據加載到內存，再在內存構建 hash table。key 為 joinkey。
- 從左表分批讀取數據，從右表 hash table匹配數據。
- 優點是：速度快 缺點是：右表數據量大的情況下占用內存。
2.3.2 Merge join?
- 對右表排序，內部 block 切分，超出內存部分 flush 到磁盤上，內存大小通過參數設定。
- 左表基于 block 排序，按照每個 block 依次與右表 merge。
- 優點是：能有效控制內存 缺點是：大數據情況下速度會慢。
優先使用 hash join 當內存達到一定閾值后再使用 merge join，優先滿足性能要求。
3. 解決方案
3.1 避免JOIN
3.1.1 數據預生成
數據預生成(由 Spark/Flink 或者 Clickhouse 物化視圖產出數據)，形成大寬表，基于單表的查詢是 ClickHouse 最為擅長的場景。
我們有個指標，實現的 SQL 比較復雜（如下），每次實時查詢很耗時，我們單獨建了一個表 table，由 Spark 每日構建出這個指標，查詢時直接基于 table 查詢。
SELECT event_date,count(distinct uc1) AS uv,sum(value) AS sum_value, ......
FROM
(SELECT event_date,hash_uid AS uc1,sum(et.float_params{ 'amount' }) AS value, count(1) AS cnt, value*cnt AS multiple
FROM tob_apps_all et GLOBAL ANY LEFT JOIN
(SELECT hash_uid AS join_key,int_profiles{ '$ab_time_34' }*1000 AS first_time
FROM users_unique_all
WHERE app_id = 10000000 AND last_active_date >= '2022-07-19' AND first_time is NOT null) upt
ON et.hash_uid=upt.join_key
WHERE (查詢條件)
GROUP BY uc1,event_date)
GROUP BY event_date;
數據量2300W，查詢時間由7秒->0.008秒。當然這種方式，需要維護額外的數據構建任務。總的思路就是不要讓 ClickHouse 實時去 JOIN。
3.1.2 使用 IN 代替 JOIN
JOIN 需要基于內存構建 hash table 且需要存儲右表全部的數據，然后再去匹配左表的數據。而 IN 查詢會對右表的全部數據構建 hash set，但是不需要匹配左表的數據，且不需要回寫數據到 block。
比如：
SELECT event_date, count()
FROM tob_apps_all et global any INNER JOIN
(SELECT hash_uid AS join_key
FROM users_unique_all
WHERE app_id = 10000000
AND last_active_date >= '2022-01-01') upt
ON et.hash_uid = upt.join_key
WHERE app_id = 10000000
AND event_date >= '2022-01-01'
AND event_date <= '2022-08-02'
GROUP BY event_date
可以改成如下形式：
SELECT event_date,
count()
FROM tob_apps_all
WHERE app_id = 10000000
AND event_date >= '2022-01-01'
AND event_date <= '2022-08-02'
AND hash_uid global IN
(SELECT hash_uid
FROM users_unique_all
WHERE (tea_app_id = 10000000)
AND (last_active_date >= '2022-01-01') )
GROUP BY event_date
如果需要從右表提取出屬性到外層進行計算，則不能使用 IN 來代替 JOIN。
相同的條件下，上面的測試 SQL，由 JOIN 時的16秒優化到了 IN 查詢時的11秒。
?
3.2 更快的 JOIN
3.2.1 優先本地 JOIN
數據預先相同規則分區
也就是 Colocate JOIN。優先將需要關聯的表按照相同的規則進行分布，查詢時就不需要分布式的 JOIN。
SELECT
et.os_name,
ut.device_id AS user_device_id
FROM tob_apps_all AS et
ANY LEFT JOIN
(
SELECT
device_id,
hash_uid
FROM users_unique_all
WHERE (tea_app_id = 268411) AND (last_active_date >= '2022-08-06')
) AS ut ON et.hash_uid = ut.hash_uid
WHERE (tea_app_id = 268411)
AND (event = 'app_launch')
AND (event_date = '2022-08-06')
settings distributed_perfect_shard=1
比如事件表 tob_apps_all 和用戶表 users_unique_all 都是按照用戶 ID 來分 shard 存儲的，相同的用戶的兩個表的數據都在同一個 shard 上，因此這兩個表的 JOIN 就不需要分布式 JOIN 了。
distributed_perfect_shard 這個 settings key 是字節內部 ClickHouse 支持的，設置過這個參數，指定執行計劃時就不會再執行分布式 JOIN 了。
基本執行過程：
- 一個 ClickHouse 節點作為 Coordinator 節點，分發查詢。在每個節點上執行 sql（tob_apps_all、users_unique_all替換成本地表）。
- 每個節點都執行1中分發的本地表 join 的 SQL（這一步不再分發右表全量的數據）。
- 數據再回傳到 coordinator 節點，然后返回給 client。
數據冗余存儲
如果一個表的數據量比較小，可以不分 shard 存儲，每個 shard 都存儲全量的數據，例如我們的業務對象表。查詢時，不需要分布式 JOIN，直接在本地進行 JOIN 即可。
SELECT count()
FROM tob_apps_all AS et
ANY LEFT JOIN
(
SELECT item_id
FROM items_all
WHERE (tea_app_id = 268411)
) AS it ON et.item_id = it.item_id
WHERE (tea_app_id = 268411)
AND (event = 'app_launch')
AND (event_date = '2022-08-06')
settings distributed_perfect_shard=1
例如這個 SQL，items_all 表每個 shard 都存儲同樣的數據，這樣也可以避免分布式 JOIN 帶來的查詢放大和全表數據分發問題。
3.2.2 更少的數據
不論是分布式 JOIN 還是本地 JOIN，都需要盡量讓少的數據參與 JOIN，既能提升查詢速度也能減少資源消耗。
SQL 下推
ClickHouse 對 SQL 的下推做的不太好，有些復雜的 SQL 下推會失效。因此，我們手動對 SQL 做了下推，目前正在測試基于查詢優化器來幫助實現下推優化，以便讓 SQL 更加簡潔。
下推的 SQL：
SELECT
et.os_name,
ut.device_id AS user_device_id
FROM tob_apps_all AS et
ANY LEFT JOIN
(
SELECT
device_id,
hash_uid
FROM users_unique_all
WHERE (tea_app_id = 268411)
AND (last_active_date >= '2022-08-06'
AND 用戶屬性條件 1 OR 用戶屬性條件 2)
) AS ut ON et.hash_uid = ut.hash_uid
WHERE (tea_app_id = 268411)
AND (event = 'app_launch')
AND (event_date = '2022-08-06')
settings distributed_perfect_shard=1
對應的不下推的 SQL：
SELECT
et.os_name,
ut.device_id AS user_device_id
FROM tob_apps_all AS et
ANY LEFT JOIN
(
SELECT
device_id,
hash_uid
FROM rangers.users_unique_all
WHERE (tea_app_id = 268411)
AND (last_active_date >= '2022-08-06')
) AS ut ON et.hash_uid = ut.hash_uid
WHERE (tea_app_id = 268411)
AND (event = 'app_launch')
AND (event_date = '2022-08-06')
AND (ut.用戶屬性條件 1 OR ut.用戶屬性條件 2)
settings distributed_perfect_shard=1
可以看到，不下推的 SQL 更加簡潔，直接基于 JOIN 過后的寬表進行過濾。但是 ClickHouse 可能會將不滿足條件的 users_unique_all 數據也進行 JOIN。
我們使用中有一個復雜的 case，用戶表過濾條件不下推有1千萬+，SQL 執行了3000秒依然執行超時，而做了下推之后60秒內就執行成功了。
3.2.3 Clickhouse 引擎層優化
一個 SQL 實際在 Clickhouse 如何執行，對 SQL 的執行時間和資源消耗至關重要。社區版的 Clickhouse 在執行模型和 SQL 優化器上還要改進的空間，尤其是復雜 SQL 以及多 JOIN 的場景下。
執行模型優化
社區版的 Clickhouse 目前還是一個兩階段執行的執行模型。第一階段，Coordinator 在收到查詢后，將請求發送給對應的 Worker 節點。第二階段，Worker 節點完成計算，Coordinator 在收到各 Worker 節點的數據后進行匯聚和處理，并將處理后的結果返回。
有以下幾個問題：
- 第二階段的計算比較復雜時，Coordinator 的節點計算壓力大，容易成為瓶頸。
- 不支持 shuffle join,hash join 時右表為大表時構建慢，容易 OOM。
- 對復雜查詢的支持不友好。
字節跳動 ClickHouse 團隊為了解決上述問題，改進了執行模型，參考其他的分布式數據庫引擎(例如 Presto 等)，將一個復雜的 Query 按數據交換情況切分成多個 Stage，各 Stage 之間則通過 Exchange 完成數據交換。根據 Stage 依賴關系定義拓撲結構，產生 DAG 圖，并根據 DAG 圖調度 Stage。例如兩表 JOIN，會先調度左右表讀取 Stage，之后再調度 JOIN 這個 Stage， JOIN 的 Stage 依賴于左右表的 Stage。
舉個例子
SELECT
et.os_name,
ut.device_id AS user_device_id,
dt.hash_did AS device_hashid
FROM tob_apps_all AS et
GLOBAL ANY LEFT JOIN
(
SELECT
device_id,
hash_uid
FROM users_unique_all
WHERE (tea_app_id = 268411) AND (last_active_date >= '2022-08-06')
) AS ut ON et.hash_uid = ut.hash_uid
GLOBAL ANY LEFT JOIN
(
SELECT
device_id,
hash_did
FROM devices_all
WHERE (tea_app_id = 268411) AND (last_active_date >= '2022-08-06')
) AS dt ON et.device_id = dt.device_id
WHERE (tea_app_id = 268411) AND (event = 'app_launch') AND (event_date = '2022-08-06')
LIMIT 10
Stage執行模型基本過程（可能的）：
- 讀取 tob_apps_all 數據，按照 join key（hash_uid）進行 shuffle，數據分發到每個節點。這是一個Stage。
- 讀取 users_unique_all 數據，按照 join key（hash_uid）進行 shuffle，數據分發到每個節點。這是一個 Stage。
- 上述兩個表的數據，在每個節點上的數據進行本地JOIN，然后再按照 join key(device_id) 進行 shuffle。這是一個 Stage。
- 讀取 devices_all 數據，按照 join key（device_id）進行 shuffle，這是一個Stage。
- 第3步、第4步的數據，相同 join key(device_id) 的數據都在同一個節點上，然后進行本地JOIN，這是一個 Stage。
- 匯總數據，返回 limit 10 的數據。這是一個 Stage。
統計效果如下：
查詢優化器
有了上面的 stage 的執行模型，可以靈活調整 SQL 的執行順序，字節跳動 Clickhouse 團隊自研了查詢優化器，根據優化規則(基于規則和代價預估)對 SQL 的執行計劃進行轉換，一個執行計劃經過優化規則后會變成另外一個執行計劃，能夠準確的選擇出一條效率最高的執行路徑，然后構建 Stage 的 DAG 圖，大幅度降低查詢時間。
下圖描述了整個查詢的執行流程，從 SQL parse 到執行期間所有內容全部進行了重新實現（其中紫色模塊），構建了一套完整的且規范的查詢優化器。
還是上面的三表 JOIN 的例子，可能的一個執行過程是：
- 查詢優化器發現 users_unique_all 表與 tob_apps_all 表的分 shard 規則一樣（基于用戶 ID ），所以就不會先對表按 join key 進行 shuffle，users_unique 與 tob_apps 直接基于本地表 JOIN，然后再按照 join key(device_id)進行 shuffle。這是一個 Stage。
- 查詢優化器根據規則或者代價預估決定設備表 devices_all 是需要 broadcast join 還是 shuffle join。
- 如果 broadcast join：在一個節點查到全部的 device 數據，然后分發到其他節點。這是一個 Stage。
- 如果 shuffle join：在每個節點對 device 數據按照 join key(device_id) 進行 shuffle。這是一個 Stage。
- 匯總數據，返回 limit 10 的數據。這是一個 Stage。
效果：
可以看到，查詢優化器能優化典型的復雜的 SQL 的執行效率，縮短執行時間。
4. 總結
ClickHouse 最為擅長的領域是一個大寬表來進行查詢，多表 JOIN 時Clickhouse 性能表現不佳。作為業內領先的用戶分析與運營平臺，火山引擎增長分析 DataFinder 基于海量數據做到了復雜指標能夠秒級查詢。本文介紹了我們是如何優化 Clickhouse JOIN 查詢的。
主要有以下幾個方面：
- 減少參與 JOIN 的表以及數據量。
- 優先使用本地 JOIN，避免分布式 JOIN 帶來的性能損耗。
- 優化本地 JOIN，優先使用內存進行 JOIN。
- 優化分布式 JOIN 的執行邏輯，依托于字節跳動對 ClickHouse 的深度定制化。
