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日志落到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.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的核心是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错误字段类型不一致显式指定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处理几十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，分区数是多少。很多性能问题在执行计划里就能看出来，不用等到跑大数据集才发现。
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日志落到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.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的核心是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错误字段类型不一致显式指定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处理几十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，分区数是多少。很多性能问题在执行计划里就能看出来，不用等到跑大数据集才发现。