1. 先想清楚Spark 读 ClickHouse 到底在解决什么问题1.1 这活儿一般出现在什么场景里Spark 读取 ClickHouse 数据这个动作本身不复杂一行spark.read.jdbc就能跑通但真正把它放进生产链路里问题就全冒出来了。我做过的几个数据平台项目里这个组合基本都出现在三类场景一类是离线数仓补数ClickHouse 里存着明细宽表需要每天凌晨把过去一天的数据抽到 Hive 或 HDFS 做长期归档和复杂关联一类是联邦分析业务侧想在 Spark 里同时拿到 ClickHouse 的高频查询结果和对象存储里的历史冷数据做一次跨源的 Join还有一类是模型训练取数算法同学要把 ClickHouse 里几百个特征字段拉成 DataFrame再喂给特征工程流程。这三类场景对读取方式的要求完全不同补数追求的是吞吐和稳定一次几十亿行也要能扛住联邦分析追求的是延迟最好只拉过滤后的几十万行模型取数追求的是列裁剪和schema 保真多拉一列就是几百 GB 的浪费类型映射错了整条训练链路全废。如果你上来就复制一段spark.read.jdbc(url, db.table, props)就往下走大概率会在半夜被告警叫醒。所以这篇文章我想按“选型、配置、踩坑、校验”这条线来讲把这件事从能跑通做到能放心跑。适合已经在用 Spark 做数据开发、但对 ClickHouse 那套存储和查询特性不够熟的同学也适合刚接手一条已有链路、想搞清楚里面每个参数为什么这么设的人。1.2 JDBC 直读、官方 Connector、导出文件中转三条路怎么选在动手写代码之前先把路选对能省掉后面一半的返工。目前主流就三条路我按实际使用感受做个对照方案实现方式优势短板适用场景Spark 内置 JDBCspark.read.jdbc clickhouse-jdbc 驱动零额外依赖Spark 自带写法最通用只能按列区间分片谓词下推弱复杂查询要手写 SQL 子查询中小数据量、临时取数、快速验证ClickHouse 官方 Spark Connectorformat(clickhouse) DataSource V2支持列裁剪、谓词下推、分区并行、读写一体版本耦合紧Spark 小版本升级要换对应 runtime 包生产链路、大数据量、长期维护导出文件中转clickhouse-client --query ... FORMAT Parquet落 HDFS再 Spark 读完全不占用 Spark 与 ClickHouse 的连接资源可断点续传多一次落地运维成本高时效性差超大批量一次性迁移、跨集群搬数我个人的取舍原则很简单数据量在千万行以内、跑一次就完事的用内置 JDBC 最省事要进调度、每天跑、数据量上亿的直接上官方 Connector只有在两边网络完全隔离、或者需要把数据搬到另一个机房的时候我才会选文件中转这条重路子。这里有个容易被忽略的点内置 JDBC 走的是 ClickHouse 的HTTP 接口默认 8123而不是原生 TCP 的 9000 端口。很多人配 URL 的时候照着客户端习惯写 9000结果一直Connection refused排查半天。记住一句话——JDBC 驱动说的是 HTTP 协议端口别写错。2. 环境准备驱动版本这一步最容易翻车2.1 驱动包与 Spark 版本的对应关系clickhouse-jdbc这些年换过好几次包名和驱动类名这是新手最容易卡住的地方。老版本用的是ru.yandex.clickhouse.ClickHouseDriver从 0.3.2 之后迁移到了com.clickhouse.jdbc.ClickHouseDriver。如果你的代码里还写死着Class.forName(ru.yandex.clickhouse.ClickHouseDriver)而 jar 包是新版的就会直接抛ClassNotFoundException。我列一下常见组合避免你在这上面浪费时间clickhouse-jdbc 版本驱动类名说明0.2.xru.yandex.clickhouse.ClickHouseDriver老包功能有限不建议新项目用0.3.2 及以上com.clickhouse.jdbc.ClickHouseDriver主流版本支持 HTTP/HTTPS0.4.x ~ 0.6.xcom.clickhouse.jdbc.ClickHouseDriver增加对all分类包的支持注意 classifier关于 classifier这里有个实际经验Maven 上clickhouse-jdbc有好几个产物clickhouse-jdbc纯 JDBC、clickhouse-jdbc-all带 shade 的全量包、还有按http、grpc分类的。在 Spark 环境里我基本只用-all那个包因为 Spark 的 classpath 里已经塞满了各种 shaded 依赖很容易和 ClickHouse 驱动内部依赖的 HTTP client 撞版本用全量 shade 包能大幅降低冲突概率。放到 Spark 提交命令里就是老老实实--jars带上或者提前扔进$SPARK_HOME/jarsspark-submit \ --master yarn \ --deploy-mode cluster \ --jars /opt/jars/clickhouse-jdbc-0.6.0-all.jar \ --conf spark.sql.shuffle.partitions400 \ your_job.py注意集群模式下--jars分发的是 driver 和 executor 都能拿到的位置但如果你在代码里用了Class.forName务必确认 executor 侧也能加载到最好在作业启动时打印一次driver.getClass.getProtectionDomain.getCodeSource把实际加载的 jar 路径打出来出问题时这是第一手证据。2.2 最小可运行配置连接参数别只写 host 和 port很多示例代码只给一个裸 URLjdbc:clickhouse://host:8123/db。能连上但跑一会儿就断。原因是 ClickHouse 的查询经常会超过默认的 socket 超时默认值在不同版本里是 30s 到 300s 不等尤其是一张几亿行的表做全表扫描的时候。我现在常用的 URL 模板长这样val url jdbc:clickhouse://ch-host:8123/analytics ?socket_timeout600000 connection_timeout30000 compress1 decompress1 use_server_time_zonefalse use_time_zoneAsia/Shanghai val props new java.util.Properties() props.setProperty(user, chUser) props.setProperty(password, chPassword) props.setProperty(driver, com.clickhouse.jdbc.ClickHouseDriver)几个参数的意义我逐个说一下。socket_timeout是网络读写超时大数据量读取必须调大我一般按“预估最慢分区耗时 × 2”来给compress/decompress打开后走 HTTP 压缩传输在网络带宽成为瓶颈时能省 50% 以上的传输时间但会略微增加 CPU实测下来跨机房场景收益非常明显同机房收益一般use_server_time_zonefalse配合use_time_zone是把时间类型的转换固定下来避免不同节点的 JVM 默认时区不一致导致同一份数据读出两个时间。提示账号密码千万别硬编码在代码里提交到仓库。我习惯用 Spark 的--conf spark.sql.catalog.xxx.password${env:CH_PASSWORD}或者在调度系统里做变量注入代码里只出现占位符。3. 核心参数拆解把单线程拉成多线程3.1 为什么默认只有一个分区在跑用spark.read.jdbc(url, db.table, props)这种最简写法Spark 拿到的 DataFrame 默认是1 个分区。也就是说不管你的集群有 200 个核都只有一个 task 在慢悠悠地从 ClickHouse 拉数据。这是新手最常见的性能问题没有之一。原因在于 Spark 的 JDBC 数据源只有在明确给出分片信息时才会切分任务。它的分片逻辑很朴素给定一个数值型的列、一个下界、一个上界和分区数它会算出每一段的区间然后给每个 task 发一条带WHERE col ? AND col ?的查询val df spark.read .format(jdbc) .option(url, url) .option(dbtable, analytics.user_events) .option(user, chUser) .option(password, chPassword) .option(partitionColumn, event_id) .option(lowerBound, 0) .option(upperBound, 100000000) .option(numPartitions, 24) .option(fetchsize, 10000) .load()这里event_id必须是数值型或可比较的列ClickHouse 里UInt64、Int64、Date转成的整数都可以但字符串列不行会报无法转换为数值。lowerBound和upperBound只是用来算区间宽度的不会真的去查最大值最小值所以它们必须你自己根据数据分布给给错了就会导致数据倾斜——比如实际数据全集中在 0 到 100 万之间你把上界设成 1 亿那 23 个分区都是空的剩下的活全压在一个 task 上。3.2 分区列怎么选才能既均匀又不给 ClickHouse 添堵选分区列我一般看两个指标取值是否连续、是否是表的主键前缀。ClickHouse 的MergeTree是按ORDER BY键做稀疏索引的如果分区列刚好是排序键的第一列那每一条分片查询都能走索引范围扫描速度快得离谱如果分区列和排序键完全无关那每条查询都会变成全表扫描24 个分区就是 24 次全表扫描反而把 ClickHouse 打爆。我踩过最惨的一次是拿一张 30 亿行的日志表用cityHash64算出来的哈希列做分区列结果每个分区都要扫全表ClickHouse 侧 CPU 直接打满影响了线上查询。后来改成用按天分区键的整数形式toYYYYMMDD(event_date)每个分片只扫对应的分区耗时从 40 分钟降到 6 分钟。所以我的建议是这样优先用日期对应的整数值做分区列天然贴合分区裁剪。如果没有日期用单调递增的 ID但要确认上下界贴近真实分布。实在没有合适的数值列就别硬用numPartitions改用官方 Connector 的partition能力或者干脆用predicates数组手动指定每一片的 WHERE 条件。predicates这个重载很容易被忽略但它是最灵活的val predicates Array( event_date 2024-01-01, event_date 2024-01-02, event_date 2024-01-03 ) val df spark.read.jdbc(url, analytics.user_events, predicates, props)每一片就是一条独立查询分片条件完全由你自己控制既能精准命中分区又能避开数值列的坑。缺点是分片数固定死在代码里调度时要能动态生成。3.3 列裁剪和谓词下推内置 JDBC 到底做了多少这是最容易被误判的地方。很多人以为 Spark 读 JDBC 时会自动只取需要的列实际上内置 JDBC 数据源在大多数情况下拉的是整行。你后面写df.select(a, b)那是在 Spark 内存里裁网络传输一点没省。对于 ClickHouse 那种一张表几百列、单行几 KB 的宽表这是巨大的浪费。想真正省传输量有两个办法。第一个是在dbtable里直接写子查询.option(dbtable, (SELECT id, user_id, event_name, event_time FROM analytics.user_events WHERE event_date 2024-01-01) t)注意必须给子查询起别名否则 Spark 拼出来的 SQL 会语法错误。这样做的好处是列裁剪和过滤都在 ClickHouse 侧完成传输量能降一个数量级。第二个办法就是用官方 Connector它原生支持把select和where下推。至于谓词下推内置 JDBC 的支持很有限基本只有where里的简单等值条件可能被下推复杂函数、in列表、日期函数经常会被拉到 Spark 侧执行。我一般的做法是把能过滤的条件全部写进子查询不依赖优化器的自动下推。这看起来笨但胜在可控。4. 官方 Connector 的实战配置与差异对比4.1 依赖引入与读写配置官方 Connector 的坐标格式是clickhouse-spark-runtime-{spark大版本}_{scala版本}比如 Spark 3.3 Scala 2.12 就是clickhouse-spark-runtime-3.3_2.12。这里有个坑Spark 小版本升级时要换包3.3 的包在 3.4 上跑大概率会报NoSuchMethodError因为 DataSource V2 的内部接口在这两个版本之间改过。配置上它走的是spark.clickhouse.*前缀spark.conf.set(spark.clickhouse.host, ch-host) spark.conf.set(spark.clickhouse.port, 8123) spark.conf.set(spark.clickhouse.protocol, http) spark.conf.set(spark.clickhouse.user, chUser) spark.conf.set(spark.clickhouse.password, chPassword) spark.conf.set(spark.clickhouse.database, analytics) val df spark.read .format(clickhouse) .option(table, user_events) .option(partition, event_date) .option(partition.num, 24) .load()它相比内置 JDBC 有几个实实在在的好处。第一列裁剪是真的做了df.select的列会体现在发给 ClickHouse 的 SELECT 子句里。第二分片不要求数值列partition可以指定日期或字符串列Connector 会自己去查min/max做切分。第三支持直接读分布式表会自动扇出到各分片并行拉取。但要注意读分布式表的时候每个 ClickHouse 分片都会承受一份查询压力如果分片数很多、Spark 又开了大并发很容易把整个集群压垮。我的做法是把partition.num控制在分片数 × 2 以内别贪多。4.2 两种方式的实测取舍我在一个中等规模的表上做过对照3000 万行、60 个字段、单机 3 节点 ClickHouseSpark 用 8 个 executor每个 4 核指标内置 JDBC子查询裁剪官方 Connector总耗时约 11 分钟约 6 分钟网络传输量约 18 GB约 7 GBClickHouse 侧峰值 CPU中等较高并发查询多代码复杂度低纯 SQL 子查询中需要配 Connector 参数版本升级风险低驱动向下兼容好高需同步换 runtime 包结论很清楚数据量越大、字段越宽Connector 的优势越明显但如果只是每天拉几百万行做个小报表内置 JDBC 加子查询完全够用还省去了版本对齐的麻烦。我现在新项目基本都用 Connector老项目能跑就不动这是实话。5. 类型映射那些不报错但结果是错的地方5.1 数值溢出、Decimal 精度和时区偏移ClickHouse 的类型系统和 Spark 不完全对齐有些映射是“能用但会悄悄出错”的这类问题最麻烦因为它不抛异常。第一个是UInt64。ClickHouse 的无符号 64 位整数上限是 2^64-1而 Spark 的LongType上限是 2^63-1。如果你的 ID 列真的用了UInt64且值超过了这个范围读进来会变成负数而且不会有任何警告。我遇到过一回用户 ID 全变成负数下游关联全空排查了一整天才定位到。解决办法是在子查询里toString(id)转成字符串读或者确认数据实际没超过 2^63 再用。第二个是Decimal。ClickHouse 的Decimal(P, S)最大精度是 76Spark 的DecimalType最大精度是 38。如果 P 超过 38读到 Spark 会截断或者报错。金额类字段一般 P 在 18 以内问题不大但如果是科学计算场景的Decimal128、Decimal256就得提前处理。第三个是DateTime64的时区。ClickHouse 存储的是不带时区的“墙上时间”加上列上声明的时区JDBC 驱动会按 URL 里的use_time_zone做转换。如果服务端列声明是Asia/Shanghai而你 URL 里写了use_time_zoneUTC读出来的时间会差 8 小时而且 Spark 里看就是个正常时间戳肉眼根本发现不了。我的经验是统一在 URL 里显式指定时区并且在数据校验环节加一条“最大时间与当前时间差距不超过 1 天”的断言能挡掉绝大多数时区错误。5.2 Array、Map、Nullable 的处理策略ClickHouse 支持嵌套类型这在日志场景里很常见。Array(String)读进 Spark 会变成ArrayType(StringType)一般没问题Map(String, String)在较新版本的驱动里会映射成MapType但老版本驱动可能直接返回字符串需要手动解析。Nullable才是真正的雷区。如果 ClickHouse 列是Nullable(String)Spark 读进来是 nullable 的没问题。但反过来如果列不是Nullable而驱动把它标记成 nullable你在做df.na.fill或者写 Hive 时不会出问题真正的问题在于空字符串和 NULL 的混淆——ClickHouse 里空字符串和 NULL 是两个值有些业务逻辑把空串当 NULL 用迁到 Spark 后行为不一致。我一般会在读取后加一层规范化处理from pyspark.sql import functions as F df df.select( *[ F.when(F.col(c) , None).otherwise(F.col(c)).alias(c) if t string else F.col(c) for c, t in df.dtypes ] )这段代码看着笨但它把“空串当空值”这个约定显式固化下来了比依赖隐含行为靠谱。6. 故障排查我总结的一份速查表6.1 报错信息与根因对照下面这些错误我几乎每个都遇到过至少一次整理成表格方便你直接对号入座报错关键字大概率根因处理办法No suitable driver found驱动类未加载或 jar 未分发到 executor检查--jars、驱动类名、是否用了 all 包Connection refused端口写成了 9000 或网络策略不通JDBC 走 HTTP确认 8123/8443 可达SocketTimeoutExceptionsocket_timeout太小或查询太重调大超时同时优化查询加分区过滤Memory limit exceededClickHouse 侧单查询超出内存配额缩小分片、加WHERE、调max_memory_usageToo many simultaneous queries并发分片数超过 ClickHouse 上限降低numPartitions或用max_concurrent_queries_for_userContainer killed by YARNexecutor 内存不足一次性拉太多行调小fetchsize、加大 executor 内存、增加分区数Cannot convert to number分区列不是数值类型换数值列或改用predicates数组Decimal precision exceeds精度超 38在子查询里toString或降低精度其中“Container killed by YARN”这条我想多说两句。它的本质是单批次拉取的行数太多堆内存撑爆。fetchsize控制的是驱动一次性从服务端缓冲区取多少行默认值在不同版本里差别很大有的版本默认是 0全部拉取那基本必炸。我现在一律显式设成 5000 到 20000 之间具体看单行宽度——如果一行有几十个字段、单行接近 1 KB那就设小一点。6.2 数据校验的三道关别只看 count数据抽完了直接看一眼行数对不对就收工这是我见过最容易埋雷的习惯。行数相同但内容错了的情况太常见了我一般走三道关第一道总量核对。在 ClickHouse 侧跑一次SELECT count() FROM ... WHERE 条件在 Spark 侧df.count()两边必须完全一致。这一步能挡住过滤条件下推失败、分区边界重复或遗漏的问题。第二道字段级抽检。在 ClickHouse 侧跑SELECT sum(amount), min(event_date), max(event_date), uniq(user_id) FROM ...Spark 侧用同样的聚合对比。这一步能挡住数值溢出、精度丢失、时区偏移这几类不报错的错误。我尤其看重sum的对比因为 Decimal 精度问题会在求和后放大。第三道抽样逐行比对。用ORDER BY cityHash64(id) LIMIT 100在两边各取 100 行做一次全字段比对。ClickHouse 的ORDER BY rand()在大表上会触发全量排序很慢用哈希函数做伪随机更稳。-- ClickHouse 侧抽样 SELECT id, user_id, amount, event_time FROM analytics.user_events WHERE event_date 2024-01-01 ORDER BY cityHash64(id) LIMIT 100这三道关做完我心里才算踏实。虽然听起来繁琐但写成脚本之后每次自动跑成本很低。7. 把一次性脚本变成可复用作业的几个心得7.1 参数化和资源配置的实操建议临时取数的脚本和进调度的作业写法的差别比想象中大。我现在的习惯是所有连接信息、表名、日期区间、分区数全部走外部参数一行硬编码都不留import argparse parser argparse.ArgumentParser() parser.add_argument(--biz_date, requiredTrue) parser.add_argument(--ch_table, requiredTrue) parser.add_argument(--partitions, typeint, default16) parser.add_argument(--fetch_size, typeint, default10000) args parser.parse_args()资源配比上我的经验值是executor 数 × 每 executor 核数 要略大于分区数否则会出现一部分 executor 空闲、一部分排队的情况。比如设了 24 个分区那就配 6 个 executor、每个 4 核刚好 24 个 slot一轮跑完。Core 数再往上加ClickHouse 那边的并发查询压力就顶不住了收益递减得非常快。另外spark.sql.shuffle.partitions这个参数在读取阶段其实不影响但如果读完之后还要做groupBy或者join默认的 200 在大数据量下会导致单个 shuffle 分区过大。我一般按“总数据量 / 单分区目标 128MB”来估算30 GB 的数据设 240 左右比较合适。7.2 几个我实际踩过的坑第一个坑是重复读。有一段代码里对同一个 DataFrame 做了两次count()看上去只是统计实际上每次 action 都会重新去 ClickHouse 拉一遍全量数据。后来我加了.persist(StorageLevel.DISK_ONLY)第二次直接从磁盘读总耗时从 50 分钟降到 28 分钟。如果你要对同一份数据做多次操作先 persist这是最便宜的优化。第二个坑是分片边界重复。Spark 的分片条件是和左闭右开所以边界值不会被重复读。但我有一次手动写predicates写成了event_date 2024-01-01和event_date 2024-01-01两片重叠数据直接翻倍。手写 predicates 的时候边界一定要严格左闭右开写完拿 count 对一遍。第三个坑是连接泄漏。在一个循环里反复创建 DataFrame 却从不unpersist跑了两百多轮之后 executor 全在 GC 上打转。现在的习惯是循环体内用完立刻df.unpersist()并且在最后统一spark.catalog.clearCache()。第四个坑是 ClickHouse 侧的负载。有一次白天跑补数24 个分区并发拉数据直接把线上的实时查询拖慢了三倍业务方找上门。从那以后我的规矩是——大批量抽取一律放在业务低峰期并且在 URL 里带上max_threads2限制单查询的并行度宁可慢一点也不能影响线上。最后分享一个我现在固定会加的小动作在作业开始时先跑一条SELECT 1做连通性探活失败就直接退出不要等到真正拉数据的时候再报错。这个看起来多余的检查帮我省过好多次“跑了两小时才发现连不上”的窝火时刻。