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

1次阅读

没有评论

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

在数据分析和机器学习领域，ClickHouse（简称 CK）作为高性能的列式数据库，常被用于存储大规模数据集。然而，当需要从远程服务器下载 CK 数据集时，开发者常遇到以下典型问题：

单线程下载速度慢：特别是对于 GB 级别的大文件，单线程下载往往成为性能瓶颈
网络波动导致失败：下载中途因网络问题中断后，需要重新开始整个下载过程
缺乏完整性校验：下载完成后难以快速验证文件是否完整无损

这些问题在大规模数据迁移或频繁更新数据集的场景下尤为突出。

多线程分块下载的核心思想是将一个大文件分割成多个小块，每个线程负责下载一个独立的块，最后合并所有块完成下载。这种方法有效利用了以下优势：

并行化 I / O 操作：现代网络接口和存储设备都能从并行访问中获益
充分利用带宽：多线程可以更好地占据可用网络带宽
细粒度控制：每个分块可以独立管理，便于实现断点续传

HTTP 协议的 Range 头允许客户端指定请求资源的特定字节范围。这是实现断点续传的基础：

服务器支持时，可以通过 Range: bytes=START-END 头请求文件的部分内容
每个下载线程维护自己的下载进度状态
中断后恢复时，只需重新请求未完成的分块

为确保下载文件的完整性，我们采用多级校验策略：

分块级校验：每个分块下载完成后立即计算并暂存其 MD5 值
全局校验：所有分块合并后，计算完整文件的 MD5 与服务器提供的校验和比对
状态持久化：将下载进度和校验信息保存到本地，支持意外中断后的快速恢复

以下是完整的 Python 实现，使用 requests 库和标准库的 threading 模块：

import os
import hashlib
import threading
from concurrent.futures import ThreadPoolExecutor

class CKDownloader:
    """
    ClickHouse 数据集多线程下载器
    支持断点续传和 MD5 校验
    """

    def __init__(self, url, target_path, chunk_size=10*1024*1020, threads=4):
        self.url = url
        self.target_path = target_path
        self.chunk_size = chunk_size  # 默认 10MB 一个分块
        self.threads = threads
        self.file_size = self._get_file_size()
        self.lock = threading.Lock()

    def _get_file_size(self):
        """获取远程文件大小"""
        with requests.head(self.url) as resp:
            return int(resp.headers.get('Content-Length', 0))

    def _download_chunk(self, chunk_id, start, end):
        """下载单个分块"""
        headers = {'Range': f'bytes={start}-{end}'}
        with requests.get(self.url, headers=headers, stream=True) as resp:
            chunk_data = resp.content
            md5 = hashlib.md5(chunk_data).hexdigest()

            # 线程安全地写入文件
            with self.lock:
                with open(self.target_path, 'r+b') as f:
                    f.seek(start)
                    f.write(chunk_data)
            return md5

    def download(self):
        """启动多线程下载"""
        # 预创建目标文件
        with open(self.target_path, 'wb') as f:
            f.truncate(self.file_size)

        # 计算分块范围
        chunks = []
        for i in range(0, self.file_size, self.chunk_size):
            end = min(i + self.chunk_size - 1, self.file_size - 1)
            chunks.append((i, end))

        # 使用线程池下载
        with ThreadPoolExecutor(max_workers=self.threads) as executor:
            futures = []
            for idx, (start, end) in enumerate(chunks):
                futures.append(executor.submit(self._download_chunk, idx, start, end))

            # 收集各分块 MD5 用于校验
            chunk_hashes = [f.result() for f in futures]

        # 完整的 MD5 校验（可选）self._verify_full_file()

    def _verify_full_file(self):
        """验证完整文件的 MD5"""
        # 实际项目中应从服务器获取预期的 MD5
        with open(self.target_path, 'rb') as f:
            md5 = hashlib.md5(f.read()).hexdigest()
            print(f'File MD5: {md5}')

我们测试了一个 2.4GB 的 CK 数据集在不同线程数下的下载表现：