1. 为什么要在Spark里同时搞定HDFS、HBase和JSON
大数据处理绕不开三件事：数据从哪来、怎么算、算完往哪放。Spark本身是个计算引擎，它不存数据，所以读写外部存储就成了日常开发里最高频的操作。我做过一个统计，一个典型的数据管道项目里，读写代码大概占30%到40%的篇幅，剩下的才是业务逻辑。换句话说，读写写不利索，后面全是坑。
HDFS、HBase、JSON这三种数据形态，基本覆盖了离线批处理场景下80%以上的需求。HDFS是底层的分布式文件系统，适合存大文件、做全量扫描；HBase是列式NoSQL数据库，适合随机读写、按rowkey快速定位；JSON则是半结构化数据的通用格式，日志、接口返回、配置文件到处都是它。Spark对这三者都有原生支持，但“支持”和“用好”之间隔着一条河。
这篇文章面向的是已经写过Spark基础代码、但对多数据源读写还不够熟练的开发者。我会从方案选型讲到参数调优，从代码模板讲到踩坑记录，尽量把每个“为什么这么写”讲清楚。读完你至少能做到：拿到一个读写需求，知道该选哪种方式、怎么配参数、遇到报错往哪个方向排查。
2. 三种数据源的特性对比与选型逻辑
2.1 HDFS、HBase、JSON各自适合什么场景
选型这件事，很多人是“手里有锤子看什么都是钉子”。我见过用HBase存全量历史数据做离线分析的，也见过用HDFS存需要频繁随机更新的维表的，结果都很痛苦。先把三者的脾气摸清楚。
HDFS的本质是一个分布式文件系统，它擅长的是顺序读写大文件。Spark读HDFS，底层走的是Hadoop的InputFormat，一次扫描一大片，吞吐量很高。它的短板是随机读写能力几乎为零，你没法高效地“更新某一行”。所以HDFS适合存原始日志、全量快照、中间结果这些“写完就不怎么改”的数据。
HBase是建立在HDFS之上的列式数据库，它的核心能力是围绕rowkey的随机读写。按rowkey查一行，毫秒级返回；按rowkey范围扫描，也能做到高效。但它的短板是分析能力弱，全表扫描性能远不如HDFS，而且不支持复杂的SQL。所以HBase适合存维度表、需要实时更新的状态数据、以及作为在线服务的底层存储。
JSON不是存储系统，是一种数据格式。它的特点是灵活、自描述、嵌套结构表达能力强。日志文件、API响应、配置数据大量使用JSON。Spark读JSON很方便，但JSON的解析开销不小，尤其是嵌套层级深、字段多的时候。所以JSON适合作为数据交换的中间格式，或者直接读取原始日志做ETL。
下面这张表是我自己总结的选型速查表，实际项目里对着看基本不会选错：
| 维度 | HDFS | HBase | JSON |
|---|---|---|---|
| 读写模式 | 顺序读写为主 | 随机读写为主 | 顺序读为主 |
| 适合数据量 | TB到PB级 | TB级 | 取决于文件大小 |
| 更新能力 | 不支持行级更新 | 支持行级更新 | 不支持 |
| 查询方式 | 全量扫描 | rowkey点查/范围查 | 全量解析 |
| Schema | 弱Schema | 列族需预定义 | 灵活Schema |
| Spark读取方式 | spark.read.text/parquet等 | spark.read.format("hbase") | spark.read.json |
| 典型场景 | 原始日志、快照 | 维表、状态数据 | 日志ETL、接口数据 |
2.2 读写方案选型背后的三个关键考量
选型不是拍脑袋，我一般会问三个问题。
第一个问题是数据要不要更新。如果数据写完就不动了，HDFS是最省心的选择，成本低、吞吐高。如果需要频繁更新某一行，比如用户画像标签每天变，那就得用HBase。JSON文件本身不支持更新，要更新只能重写整个文件，所以它不适合做需要更新的存储。
第二个问题是查询模式是什么。如果是全量分析，比如“统计昨天所有订单的金额分布”，HDFS加Parquet列式存储是最优解。如果是“查某个用户最近10条记录”，HBase的rowkey设计好了就是毫秒级。JSON适合那种“先把原始数据读进来，再在Spark里做各种转换”的场景。
第三个问题是数据量和成本。HDFS存PB级数据成本可控，HBase因为要维持RegionServer和内存缓存，单位存储成本更高。JSON文件如果不大，直接放HDFS上读也没问题；如果JSON文件特别多特别小，那就要先做小文件合并，否则Spark会启动大量task，调度开销吃掉大部分时间。
提示：实际项目里这三种往往不是三选一，而是组合使用。常见架构是：原始JSON日志落到HDFS，ETL后写入HBase供在线查询，同时把聚合结果以Parquet格式写回HDFS供离线分析。
3. Spark读写HDFS的核心细节与实操
3.1 读HDFS：从text到parquet的格式选择
Spark读HDFS，最常用的入口是spark.read。但读什么格式，差别很大。
读纯文本用spark.read.text()，它把每行作为一个字符串，返回DataFrame只有一列value。这个方式简单，但后续要自己解析，适合非结构化日志的初步读取。
读CSV用spark.read.csv()，要指定header、sep、inferSchema等参数。这里有个坑：inferSchema=true会额外扫描一遍数据来推断类型，数据量大时开销明显。生产环境我一般显式指定schema，既快又稳。
读Parquet用spark.read.parquet()，这是我最推荐的离线分析格式。Parquet是列式存储，自带schema，压缩比高，Spark读它还能做谓词下推和列裁剪。同样一份数据，Parquet的读取速度通常是CSV的3到5倍。
读ORC用spark.read.orc()，和Parquet类似，在Hive生态里更常见。
代码上大概是这样：
# 读文本 df_text = spark.read.text("/data/logs/2024-01-01/") # 读CSV，显式指定schema from pyspark.sql.types import StructType, StructField, StringType, IntegerType schema = StructType([ StructField("user_id", StringType(), True), StructField("amount", IntegerType(), True), StructField("ts", StringType(), True) ]) df_csv = spark.read.option("header", "true").schema(schema).csv("/data/orders/") # 读Parquet df_parquet = spark.read.parquet("/data/warehouse/orders_parquet/")读HDFS时有个参数值得关注：spark.sql.files.maxPartitionBytes，默认128MB。它决定了一个分区最多读多少字节。如果文件很大但分区数太少，可以调小这个值来增加并行度；反之如果分区太多task调度开销大，可以调大。
3.2 写HDFS：分区、压缩与文件大小控制
写HDFS比读更容易出问题，因为写出去的文件布局直接影响下游读取效率。
写Parquet用df.write.parquet()，写CSV用df.write.csv()。但直接写往往不够，要控制几个东西。
分区是第一个要控制的。df.write.partitionBy("dt", "region")会按指定列做目录分区，下游按分区过滤时能大幅减少扫描量。但分区列不能选基数太高的，比如user_id有几百万个值，按它分区会产生几百万个目录，NameNode直接爆炸。一般分区列的基数控制在几百到几千比较合适。
文件大小是第二个要控制的。Spark写HDFS时，每个分区对应一个输出文件。如果DataFrame有200个分区，就会写200个文件。如果每个文件只有几KB，那就是典型的小文件问题。解决办法是在写之前用df.repartition(n)或df.coalesce(n)调整分区数。经验值是每个文件128MB到256MB比较合适。
# 按分区列写入，控制文件数 df.repartition(50).write \ .mode("overwrite") \ .partitionBy("dt") \ .parquet("/data/warehouse/orders_parquet/")压缩是第三个要控制的。Parquet默认用Snappy压缩，平衡了压缩率和CPU开销。如果存储紧张可以用GZIP，压缩率更高但更慢。设置方式是spark.conf.set("spark.sql.parquet.compression.codec", "gzip")。
注意：
mode("overwrite")会先删除目标路径再写。如果目标路径下有其他数据，会被一起删掉。生产环境慎用，或者确保路径是独立的。
3.3 HDFS读写的性能调优参数
调优这件事，参数很多，但真正影响大的就那么几个。我列一下自己常用的：
| 参数 | 默认值 | 建议值 | 作用 |
|---|---|---|---|
| spark.sql.files.maxPartitionBytes | 128MB | 128-256MB | 单分区最大读取字节数 |
| spark.sql.files.openCostInBytes | 4MB | 4-8MB | 打开文件的估算开销 |
| spark.sql.shuffle.partitions | 200 | 按数据量调 | shuffle后分区数 |
| spark.hadoop.mapreduce.input.fileinputformat.split.maxsize | 无 | 256MB | 控制split大小 |
| spark.sql.parquet.compression.codec | snappy | snappy/gzip | Parquet压缩算法 |
spark.sql.shuffle.partitions这个参数特别重要。默认200，数据量小的时候200个分区每个都很小，task启动开销占比高；数据量大的时候200个分区每个都很大，容易OOM。我一般按“数据量除以128MB”来估算，比如10GB数据就设80左右。
还有一个容易忽略的点：读HDFS时如果文件数远大于分区数，Spark会把多个文件合并到一个分区里读，这是好事。但如果文件数特别多（比如几十万个小文件），即使合并，NameNode的元数据压力也很大。这种情况建议先用一个单独的Job做小文件合并，再跑主逻辑。
4. Spark读写HBase的实战要点
4.1 HBase的rowkey设计与Spark读取的配合
HBase的核心是rowkey，rowkey设计不好，Spark读起来就是灾难。
rowkey设计有三个原则：长度适中、散列分布、前缀可扫描。长度一般控制在10到100字节，太短容易冲突，太长浪费存储。散列分布是为了避免热点，比如用user_id做rowkey时，如果user_id是递增的，写入会集中在最后一个Region。解决办法是加盐或者用hash前缀。前缀可扫描是指把常用的查询条件放在rowkey前面，比如“查某个用户某天的记录”，rowkey设计成user_id + date + timestamp。
Spark读HBase，官方没有内置的DataFrame接口，需要用spark.read.format("hbase")配合第三方连接器，或者用newAPIHadoopRDD自己封装。我一般用HBase社区维护的hbase-spark连接器，它提供了HBaseContext，可以批量读写。
读的时候要指定scan的范围。如果全表扫描，一定要设setCaching和setBatch。setCaching控制一次RPC取多少行，默认太小会导致RPC次数过多；setBatch控制一次取多少列，列多的时候要调大。
# 用HBaseContext读取 hbaseConf = {"hbase.zookeeper.quorum": "zk1,zk2,zk3", "hbase.zookeeper.property.clientPort": "2181"} hbaseContext = HBaseContext(spark.sparkContext, hbaseConf) # 定义scan scan = Scan() scan.setStartRow(Bytes.toBytes("user_0001")) scan.setStopRow(Bytes.toBytes("user_0100")) scan.setCaching(1000) scan.setBatch(100) rdd = hbaseContext.hbaseRDD(TableName.valueOf("user_table"), scan)4.2 批量写入HBase的BufferedMutator用法
写HBase比读更容易出性能问题。逐条Put是最慢的，因为每条都要一次RPC。正确做法是用BufferedMutator做批量写。
BufferedMutator会先把Put缓存在客户端，达到阈值后一次性发给RegionServer。关键参数是writeBufferSize，默认2MB，可以调到8MB到16MB。但也不能太大，太大一旦失败重试的数据量就大。
在Spark里，通常是对每个分区用一个BufferedMutator，分区内遍历写。这样既利用了批量，又避免了跨分区共享连接。
def write_to_hbase(partition): conf = HBaseConfiguration.create() conf.set("hbase.zookeeper.quorum", "zk1,zk2,zk3") conn = ConnectionFactory.createConnection(conf) table = conn.getTable(TableName.valueOf("user_table")) mutator = conn.getBufferedMutator(TableName.valueOf("user_table")) for row in partition: put = Put(Bytes.toBytes(row.user_id)) put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(row.name)) mutator.mutate(put) mutator.flush() mutator.close() conn.close() df.foreachPartition(write_to_hbase)注意：
foreachPartition里创建的连接是每个分区一个，分区数多的时候连接数会很多。要确保HBase集群能承受，或者用连接池。另外mutator.flush()一定要调，否则最后一批数据可能没发出去。
4.3 HBase读写常见报错与排查
HBase相关的报错，我遇到最多的有这么几类。
RegionTooBusyException：通常是写入太猛，单个Region的memstore满了。解决办法是降低写入速率，或者预分区把数据打散。
RetriesExhaustedWithDetailsException：重试次数用完了。要看具体原因，可能是RegionServer挂了，也可能是rowkey热点导致某个Region过载。日志里会列出失败的row，根据row分析。
ScannerTimeoutException：scan超时。调大hbase.client.scanner.timeout.period，或者减小setCaching让每次RPC数据量小一点。
OutOfMemoryError：客户端内存不够。检查writeBufferSize是不是设太大了，或者一次scan的数据量太大。
排查思路是：先看HBase客户端日志，再看RegionServer日志，最后看监控指标（Region的读写QPS、memstore大小、GC情况）。
5. Spark处理JSON数据的完整流程
5.1 读JSON：schema推断与显式指定
spark.read.json()是最直接的入口。它支持单个文件、目录、通配符。
默认情况下，Spark会采样一部分数据推断schema。这个采样比例由samplingRatio控制，默认1.0，也就是全量扫描。数据量大时建议调小，比如0.1，但代价是可能推断错类型。生产环境我强烈建议显式指定schema，原因有两个：一是避免推断开销，二是避免因为某条脏数据导致整个schema推断错误。
from pyspark.sql.types import StructType, StructField, StringType, LongType, ArrayType schema = StructType([ StructField("order_id", StringType(), True), StructField("user_id", StringType(), True), StructField("amount", LongType(), True), StructField("items", ArrayType(StructType([ StructField("sku", StringType(), True), StructField("qty", LongType(), True) ])), True) ]) df = spark.read.schema(schema).json("/data/logs/orders/")JSON的嵌套结构在DataFrame里会变成嵌套的StructType或ArrayType。访问嵌套字段用df.select("items.sku")或者df.selectExpr("explode(items) as item")。
5.2 处理嵌套JSON与数组展开
嵌套JSON是处理起来最烦的部分。常见操作有这么几种。
取嵌套字段：df.select(col("address.city"))，如果中间有null，结果是null，不会报错。
展开数组：用explode把数组拆成多行。比如一个订单有多个商品，explode(items)后每个商品一行。如果数组可能为空但不想丢数据，用explode_outer。
展开结构体：用df.select("address.*")把结构体里的字段平铺成多列。
处理深层嵌套：有时候JSON层级很深，比如a.b.c.d。可以逐层select，也可以写一个递归函数自动展开。我一般用df.selectExpr("a.b.c.d as d")这种写法，简洁。
from pyspark.sql.functions import explode, col # 展开items数组 df_exploded = df.select( col("order_id"), col("user_id"), explode(col("items")).alias("item") ).select( col("order_id"), col("user_id"), col("item.sku").alias("sku"), col("item.qty").alias("qty") )提示：
explode之后数据量会膨胀，如果原数组很大，要注意内存。可以在explode之前先过滤掉不需要的行。
5.3 写JSON与格式控制
写JSON用df.write.json()。和写HDFS类似，要控制分区数和文件大小。
JSON写出去默认是每行一个JSON对象，这叫JSON Lines格式，也是Spark读JSON时期望的格式。如果写成格式化的多行JSON，Spark读的时候会出问题。
写JSON时有个参数ignoreNullFields，默认true，会忽略值为null的字段。如果下游需要保留null字段，要设成false。
df.write \ .mode("overwrite") \ .option("ignoreNullFields", "false") \ .json("/data/output/orders_json/")JSON的压缩可以用compression参数，支持gzip、bzip2等。但JSON本身压缩比不如Parquet，如果只是中间结果，建议还是用Parquet。
6. 三种数据源混合读写的典型架构
6.1 从JSON到HDFS再到HBase的完整链路
实际项目里最常见的链路是：原始数据是JSON日志，先落到HDFS，然后Spark读JSON做ETL，清洗后的明细写回HDFS（Parquet格式），同时把需要在线查询的维度数据写入HBase。
这个链路里，JSON到Parquet的转换是核心。JSON灵活但读取慢，Parquet列式存储读取快。转换时要注意schema对齐，JSON里可能有些字段时有时无，Parquet要求schema固定，所以要在ETL时把缺失字段补上默认值。
写入HBase的部分，通常是维表数据。比如用户维度，从JSON日志里提取user_id和对应的属性，写入HBase。这里要注意rowkey设计，一般用user_id直接做rowkey，如果user_id分布不均，加个hash前缀。
6.2 读写过程中的数据一致性保障
跨数据源写入，一致性是个大问题。比如Spark同时写HDFS和HBase，如果HDFS写成功了HBase写失败了，数据就不一致。
保障一致性有几个思路。两阶段提交是最严格的，但Spark原生不支持，要自己实现，复杂度高。幂等写入是更实用的方案，让每次写入都是可重复的，失败了重跑不会产生重复数据。HBase的Put本身是幂等的，同样的rowkey再写一次会覆盖。HDFS写入可以用mode("overwrite")保证幂等，但要注意overwrite会删掉整个目录。
先写HDFS再写HBase是常见的顺序，因为HDFS写入相对稳定，HBase可能因为Region问题失败。如果HBase写失败，可以只重跑HBase写入部分，HDFS不用动。
注意：如果ETL逻辑里有聚合，重跑时要注意聚合的幂等性。比如count类型的聚合，重跑会翻倍。这种情况要么用overwrite，要么在聚合前先按业务键去重。
7. 常见问题与排查技巧实录
7.1 读写HDFS的典型报错
FileAlreadyExistsException：写HDFS时目标路径已存在。用mode("overwrite")或mode("append")解决。但append在Parquet上支持不好，建议用overwrite。
TooManyOpenFiles：打开文件数超过系统限制。原因是分区数太多，每个分区都要打开文件。解决办法是减少分区数，或者调大ulimit -n。
java.io.IOException: No space left on device：磁盘满了。检查HDFS剩余空间，或者清理临时目录。
Task not serializable：在闭包里引用了不可序列化的对象。比如在foreachPartition里引用了外部的Connection。解决办法是把Connection的创建放到partition内部。
7.2 HBase连接与超时问题
Connection refused：连不上ZooKeeper或RegionServer。检查网络、端口、防火墙。
SocketTimeoutException：RPC超时。调大hbase.rpc.timeout和hbase.client.operation.timeout。
RegionServer宕机导致写入失败：HBase客户端会自动重试，但如果重试次数用完还是失败，就要检查RegionServer状态。常见原因是RegionServer的GC时间过长，或者memstore满了。
ZooKeeper session expired：ZooKeeper会话超时。调大zookeeper.session.timeout，默认180秒，可以调到300秒。
7.3 JSON解析的坑与解决方案
字段类型不一致：同一个字段，有的记录是数字，有的是字符串。Spark推断schema时会报错或者推断成String。解决办法是显式指定schema，或者在读取时用mode("PERMISSIVE")让Spark把无法解析的字段设为null。
JSON格式错误：某行不是合法JSON。Spark默认会抛异常。可以用mode("DROPMALFORMED")丢弃坏行，或者mode("PERMISSIVE")保留坏行但字段为null。
嵌套层级太深：Spark对嵌套深度有限制，太深会栈溢出。解决办法是先用文本方式读进来，自己用JSON库解析，或者用from_json函数手动处理。
大JSON文件：单个JSON文件几个GB，Spark读的时候一个task处理整个文件，容易OOM。解决办法是先把大文件拆成小文件，或者用spark.read.text()读进来后自己解析。
下面这张表是我整理的常见问题速查：
| 问题 | 可能原因 | 解决方案 |
|---|---|---|
| HDFS写入文件数过多 | 分区数太多 | repartition/coalesce减少分区 |
| HBase写入慢 | 逐条Put | 用BufferedMutator批量写 |
| JSON读取schema错误 | 字段类型不一致 | 显式指定schema |
| HBase scan超时 | caching太小 | 调大setCaching |
| HDFS读取OOM | 单分区数据太大 | 调小maxPartitionBytes |
| JSON解析丢数据 | 坏行被丢弃 | 用PERMISSIVE模式保留 |
8. 我在实际项目里踩过的坑和总结的经验
说几个印象深刻的坑。
第一个是HBase写入时用了foreach而不是foreachPartition。foreach是每行都创建一次连接，几百万行就是几百万次连接创建，直接把ZooKeeper打挂了。改成foreachPartition后，每个分区一个连接，性能提升了几十倍。这个坑的本质是没理解Spark的分布式执行模型，Driver端的代码和Executor端的代码要分清楚。
第二个是JSON读取时没指定schema，让Spark自己推断。结果某天上游加了一个字段，类型和之前不一致，Spark推断出来的schema变了，下游的Parquet写入直接失败。后来改成显式schema，并且加了schema版本管理，上游改字段要先通知。
第三个是HDFS写入时分区数设成了1，想着“数据量不大一个文件就行”。结果数据量涨上来后，单个task处理几十GB，OOM了。后来改成按数据量动态计算分区数，公式是math.ceil(total_size / 128MB)。
第四个是HBase的rowkey设计。一开始用时间戳做rowkey，写入全集中在最后一个Region，热点严重。后来改成hash(user_id) + user_id + timestamp，写入打散了，但读取时因为hash前缀，没法按user_id范围扫描。最后折中成user_id + timestamp，并在user_id前面加了一位散列前缀，既打散了写入，又保留了前缀扫描能力。
这些坑的共同点是：都不是代码语法错误，而是对数据特性和系统特性的理解不到位。Spark的API很简单，难的是知道在什么场景下用什么参数、什么设计。
最后分享一个实用技巧：读写代码写完后，先用小数据集跑一遍，看执行计划。df.explain()能看到Spark实际会怎么执行，有没有多余的shuffle，分区数是多少。很多性能问题在执行计划里就能看出来，不用等到跑大数据集才发现。
