使用Spark(Scala)读取ClickHouse超大表写入HDFS的切分方案及异常处理

按1个月时间周期分片触发java.io.EOFException: reached end of stream after reading 1572 bytes; 23242 bytes expected异常，核心是两个问题：

1. 单分片数据量过大：3000亿行规模的表单月数据量普遍在百亿级，单Task拉取时单次结果集过大，要么超过ClickHouse服务端单查询返回阈值，要么传输超时被服务端主动断开连接，才会出现流读到一半中断的EOF报错。
2. 切分维度不合理：仅靠时间维度粗粒度切分容易出现数据倾斜，业务高峰月份/天的数据量可能是平峰的数倍，部分Task负载远超处理能力，同时没有利用ClickHouse原生分片能力，拉取效率低稳定性差。

• 优先用ClickHouse原生能力做均匀切分，放弃单月粗粒度分片
  首先把第一层切分粒度降到天级，匹配ClickHouse表的原生分区键（绝大多数亿级以上规模的CH表都会按天/小时建分区），不要跨整月拉取。如果天级数据量仍超过单Task合理处理阈值（单Task处理300w-800w行稳定性最高，可根据行宽调整），叠加CH的SAMPLE键二次切分：如果建表时配置了SAMPLE BY高基数字段（如用户ID哈希、自增主键），直接通过Spark ClickHouse Connector配置自动切分，每个分片查询自动携带SAMPLE子句，保证所有分片数据量均匀无倾斜。
  参考Scala读取配置：

  val chDF = spark.read
    .format("clickhouse")
    .option("url", "jdbc:clickhouse://<ch-address>:8123/<db>")
    .option("user", "<username>")
    .option("password", "<password>")
    .option("dbtable", "<target_table>")
    // 基于CH内置隐藏分区字段做第一层切分，无需手动硬编码时间范围
    .option("partitionColumn", "__partition_id")
    // 按总数据量配置分区数：3000亿行按单分片400w行算，配置750左右即可，可根据集群资源上下调整
    .option("numPartitions", "800")
    // 开启采样切分（需表提前配置SAMPLE BY字段）
    .option("useSample", "true")
    .option("sampleColumn", "<your_sample_field>")
    // 连接层配置避免流中断
    .option("socketTimeout", "300000")
    .option("fetchSize", "100000")
    .option("compress", "true")
    .load()
  

• 连接层参数调优，从根源避免传输中断
  大部分EOF报错和传输配置不合理直接相关，调整以下配置即可覆盖90%以上的流断场景：
  • JDBC连接串追加参数max_result_bytes=0&socket_timeout=300000，关闭单查询结果集大小限制，将Socket超时调整为5分钟，适配大结果集拉取场景
  • 开启LZ4传输压缩，可减少70%左右的传输数据量，降低网络传输超时概率
  • ClickHouse服务端提前调大max_concurrent_queries参数，匹配Spark并发Task数，避免连接数超限被服务端主动断连
• 写入HDFS配套优化，避免下游反压牵连读取链路
  读取后直接写入HDFS，不要额外做全量Shuffle类操作（如全局排序、重分区到极少分区），避免CH连接长时间空闲超时断开。写入优先选列式存储格式，参考配置：

  chDF.write
    .mode("append")
    .format("parquet")
    .option("compression", "zstd")
    // 和CH分区对齐，按天写HDFS分区，后续查询效率更高
    .partitionBy("dt")
    .save("<hdfs-target-path>")
  

如果目标CH表建表时未配置SAMPLE BY字段、无法修改表结构，就手动做双层切分：第一层按天拆分，第二层对表内高基数字段（如用户ID、订单ID）做哈希取模拆分，单分片数据量控制在500w行以内即可。每个分片查询条件追加and cityHash64(<high_cardinality_field>) % <shard_num_per_day> = <current_shard_id>，保证分片数据均匀。

注意：禁止用LIMIT + OFFSET的方式做分页切分，该方式下越靠后的分页查询越需要CH扫描全量数据，会大幅加重服务端负载，反而更容易触发超时断连。

内容的提问来源于stack exchange，提问作者Archi