ClickHouse数据集高效下载方案：ck数据集下载优化实战

1次阅读

没有评论

共计 2229 个字符，预计需要花费 6 分钟才能阅读完成。

在实际使用 ClickHouse（简称 CK）时，数据集下载常遇到三个主要问题：

单线程下载速度慢：特别是 GB 级以上的数据集，线性下载耗时过长
网络波动导致失败：跨国传输或弱网环境下，长连接极易中断
完整性验证缺失：下载完成后才发现数据损坏，需重新下载

通过 HEAD 请求获取文件总大小
将文件划分为 N 个等大的块（除最后一块）
每个线程负责下载指定字节范围的数据
最后合并所有分块文件

关键优势：

充分利用带宽资源
规避 TCP 慢启动问题
各分块可独立重试

基于 HTTP Range 头实现：

请求时携带 Range: bytes=start-end 头
服务端返回 206 Partial Content 状态码
本地记录已下载的块序号
中断后跳过已完成块

下载前检查本地是否存在.part 临时文件
完成下载后验证：
文件大小匹配
MD5 校验和一致
校验失败自动触发重新下载

import requests
import threading
import hashlib
import os
from pathlib import Path

class CKDownloader:
    """
    ClickHouse 数据集多线程下载器
    特性：- 多线程分块下载
    - 断点续传支持
    - MD5 完整性校验
    """

    def __init__(self, url, threads=8, chunk_size=10*1024*1024):
        self.url = url
        self.threads = threads
        self.chunk_size = chunk_size
        self.temp_dir = Path('download_temp')
        self.temp_dir.mkdir(exist_ok=True)

    def get_file_size(self):
        """通过 HEAD 请求获取文件总大小"""
        resp = requests.head(self.url)
        return int(resp.headers['Content-Length'])

    def download_chunk(self, chunk_id, start_byte, end_byte):
        """下载单个数据块"""
        temp_file = self.temp_dir / f'chunk_{chunk_id}.part'
        if temp_file.exists():
            return  # 跳过已下载块

        headers = {'Range': f'bytes={start_byte}-{end_byte}'}
        resp = requests.get(self.url, headers=headers, stream=True)
        with open(temp_file, 'wb') as f:
            for chunk in resp.iter_content(8192):
                f.write(chunk)

    def verify_integrity(self, target_path):
        """MD5 校验文件完整性"""
        # 实现校验逻辑（示例）return True

    def download(self, save_path):
        """主下载方法"""
        file_size = self.get_file_size()
        chunks = [(i, i*self.chunk_size, min((i+1)*self.chunk_size-1, file_size-1)) 
                 for i in range(0, self.threads)]

        # 多线程下载
        threads = []
        for chunk in chunks:
            t = threading.Thread(target=self.download_chunk, args=chunk)
            threads.append(t)
            t.start()

        for t in threads:
            t.join()

        # 合并文件
        with open(save_path, 'wb') as outfile:
            for i in range(self.threads):
                chunk_file = self.temp_dir / f'chunk_{i}.part'
                with open(chunk_file, 'rb') as infile:
                    outfile.write(infile.read())
                chunk_file.unlink()

        if not self.verify_integrity(save_path):
            os.remove(save_path)
            raise ValueError("文件校验失败")

        return True

测试环境：AWS EC2 t3.xlarge (4vCPU), 1GB 测试文件