全栈攻城狮

Appearance

Spark 读写外部数据源 ​

概述 ​

Spark 通过 DataSource API 统一读写外部系统。数据源选型与参数配置直接决定 ETL 性能：JDBC 要控分区避免压垮数据库，Kafka 要调吞吐，HBase/Redis 要防热点。本文逐个讲解六类常用数据源的读写方式与调优要点。

一、数据源抽象 ​

1.1 统一接口 ​

scala
spark.read.format("jdbc")
  .option("url", ...).option("dbtable", "orders").load()

df.write.format("parquet").save("hdfs:///out")
方法说明
format数据源类型
option数据源参数
load/save读写入口
jdbc/parquet/...便捷方法

1.2 分区与并行度 ​

概念说明
分区数读数据源的 Task 数
分区策略按列范围 / 按分区键
数据本地性尽量就近读

二、JDBC（MySQL/PostgreSQL/Oracle） ​

2.1 读 ​

scala
spark.read.jdbc(
  url = "jdbc:mysql://host:3306/db",
  table = "orders",
  column = "id",                 // 分区列
  lowerBound = 1, upperBound = 10000000,
  numPartitions = 10,
  connectionProperties = props
)

2.2 分区调优 ​

参数说明
partitionColumn分区列（需有序，如主键）
lowerBound/upperBound分区范围
numPartitions分区数（并行度）
fetchsize每次拉取行数
queryTimeout查询超时
风险处理
并发太高压垮数据库numPartitions 控制并发度（≤10-20）
分区列分布不均选均匀列或自定义查询
大表全量拉取按时间增量同步

2.3 写 ​

scala
df.write.mode("append").option("batchsize", "1000").jdbc(url, "orders", props)
注意说明
batchsize批量提交（默认 1000）
主键冲突用 upsert 或先删后插
写压力控制写入并行度

三、Hive ​

3.1 读写 ​

scala
spark.read.table("dw.ods_orders")     // 读 Hive 表
df.write.saveAsTable("dw.dws_report")  // 写 Hive 表
说明值
分区表自动识别分区列
动态分区spark.sql.sources.partitionOverwriteMode 控制覆盖
格式默认 Parquet

3.2 调优 ​

参数作用
spark.sql.hive.convertMetastoreParquet用原生 Parquet 读
向量化读取大表读加速
分区裁剪只读必要分区

四、HBase ​

4.1 读 ​

scala
spark.read.format("org.apache.hadoop.hbase.spark")
  .option("hbase.table", "user")
  .option("hbase.columns.mapping", "id f:name, f:age")
  .load()

4.2 要点 ​

要点说明
RowKey 扫描读时用 RowKey 范围（startRow/stopRow）
列族映射指定 family:qualifier
缓存HBase 端 blockcache 调优
避免全表扫描大表全扫极慢，用过滤器

4.3 调优 ​

手段说明
增大 scan 缓存setCaching 减少 RPC
并行 Region 扫描分区数与 Region 数匹配
批量写用 HFile 批量生成或 Put 批量

五、Kafka ​

5.1 读（批量） ​

scala
spark.read.format("kafka")
  .option("kafka.bootstrap.servers", "kafka:9092")
  .option("subscribe", "topic")
  .option("startingOffsets", "earliest")
  .option("endingOffsets", "latest")
  .load()
  .selectExpr("CAST(value AS STRING)")

5.2 吞吐调优 ​

参数说明
minPartitions最小分区数（≥ Topic 分区数）
maxOffsetsPerTrigger每次读取上限（批量场景）
fetchOffset.numRetries拉 offset 重试
kafka.batch.size批量拉取大小
场景配置方向
一次性全量读minPartitions 拉满，maxOffsetsPerTrigger 置大
流式增量Structured Streaming 用触发器控制

六、Elasticsearch ​

6.1 读写 ​

scala
spark.read.format("org.elasticsearch.spark.sql")
  .option("es.nodes", "es:9200")
  .option("es.resource", "index/doc")
  .option("es.query", """{"match_all":{}}""")
  .load()

df.write.format("org.elasticsearch.spark.sql")
  .option("es.resource", "index")
  .option("es.mapping.id", "id")
  .mode("append").save()

6.2 调优 ​

参数作用
es.batch.size.entries批量写条目数
es.batch.size.bytes批量写字节数
es.index.read.missing.as.empty缺索引容错
es.scroll.size滚动读大小
es.nodes.wan.only跨网络访问
注意说明
大批量写压垮 ES控制分区与批量大小
映射冲突写前保证 mapping 兼容
深分页用 scroll 而非深分页查询

七、Redis ​

7.1 读 ​

scala
spark.read.format("org.apache.spark.sql.redis")
  .option("host", "redis-host")
  .option("port", "6379")
  .option("table", "user_cache")
  .load()

7.2 要点 ​

要点说明
key 设计按前缀/模式读取
热点 key读压集中导致 Redis 热点
pipeline批量读用 pipeline
写入模式批量 set/批量写 hash
风险处理
大 key拆分或换存储
热点加缓存分层、随机后缀
连接数爆炸控制并行度，用连接池

八、数据源选型对照 ​

数据源典型用途读优化写优化
JDBC业务库同步分区列+增量批量提交
Hive数仓主存储分区裁剪+向量化saveAsTable
HBase实时 KVRowKey 扫描批量 Put
Kafka消息源分区对齐批量+ack
ES检索scroll批量+id
Redis缓存/维表pipeline批量

常见问题速查 ​

问题原因与处理
JDBC 拉取压垮库减少分区与 fetchsize，走增量
Kafka 读太快积压用 maxOffsetsPerTrigger 限速
ES 写超时降批量与并行，调 es 集群
HBase 全表扫慢加 RowKey 范围过滤
Redis 连接过多控制并行度、复用连接