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。下面这张表是我自己总结的选型速查表实际项目里对着看基本不会选错维度HDFSHBaseJSON读写模式顺序读写为主随机读写为主顺序读为主适合数据量TB到PB级TB级取决于文件大小更新能力不支持行级更新支持行级更新不支持查询方式全量扫描rowkey点查/范围查全量解析Schema弱Schema列族需预定义灵活SchemaSpark读取方式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日志落到HDFSETL后写入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等参数。这里有个坑inferSchematrue会额外扫描一遍数据来推断类型数据量大时开销明显。生产环境我一般显式指定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.maxPartitionBytes128MB128-256MB单分区最大读取字节数spark.sql.files.openCostInBytes4MB4-8MB打开文件的估算开销spark.sql.shuffle.partitions200按数据量调shuffle后分区数spark.hadoop.mapreduce.input.fileinputformat.split.maxsize无256MB控制split大小spark.sql.parquet.compression.codecsnappysnappy/gzipParquet压缩算法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的核心是rowkeyrowkey设计不好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分析。ScannerTimeoutExceptionscan超时。调大hbase.client.scanner.timeout.period或者减小setCaching让每次RPC数据量小一点。OutOfMemoryError客户端内存不够。检查writeBufferSize是不是设太大了或者一次scan的数据量太大。排查思路是先看HBase客户端日志再看RegionServer日志最后看监控指标Region的读写QPS、memstore大小、GC情况。5. Spark处理JSON数据的完整流程5.1 读JSONschema推断与显式指定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时期望的格式。如果写成格式化的多行JSONSpark读的时候会出问题。写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清洗后的明细写回HDFSParquet格式同时把需要在线查询的维度数据写入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。检查网络、端口、防火墙。SocketTimeoutExceptionRPC超时。调大hbase.rpc.timeout和hbase.client.operation.timeout。RegionServer宕机导致写入失败HBase客户端会自动重试但如果重试次数用完还是失败就要检查RegionServer状态。常见原因是RegionServer的GC时间过长或者memstore满了。ZooKeeper session expiredZooKeeper会话超时。调大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文件几个GBSpark读的时候一个task处理整个文件容易OOM。解决办法是先把大文件拆成小文件或者用spark.read.text()读进来后自己解析。下面这张表是我整理的常见问题速查问题可能原因解决方案HDFS写入文件数过多分区数太多repartition/coalesce减少分区HBase写入慢逐条Put用BufferedMutator批量写JSON读取schema错误字段类型不一致显式指定schemaHBase scan超时caching太小调大setCachingHDFS读取OOM单分区数据太大调小maxPartitionBytesJSON解析丢数据坏行被丢弃用PERMISSIVE模式保留8. 我在实际项目里踩过的坑和总结的经验说几个印象深刻的坑。第一个是HBase写入时用了foreach而不是foreachPartition。foreach是每行都创建一次连接几百万行就是几百万次连接创建直接把ZooKeeper打挂了。改成foreachPartition后每个分区一个连接性能提升了几十倍。这个坑的本质是没理解Spark的分布式执行模型Driver端的代码和Executor端的代码要分清楚。第二个是JSON读取时没指定schema让Spark自己推断。结果某天上游加了一个字段类型和之前不一致Spark推断出来的schema变了下游的Parquet写入直接失败。后来改成显式schema并且加了schema版本管理上游改字段要先通知。第三个是HDFS写入时分区数设成了1想着“数据量不大一个文件就行”。结果数据量涨上来后单个task处理几十GBOOM了。后来改成按数据量动态计算分区数公式是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分区数是多少。很多性能问题在执行计划里就能看出来不用等到跑大数据集才发现。