Spark SQL 提供了丰富的外部数据源集成能力，支持多种数据格式和存储系统。

数据源集成架构

JDBC 数据源集成

读取JDBC数据源

基本连接配置

// Scala - 读取JDBC表
val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/mydb")
  .option("dbtable", "sales")
  .option("user", "username")
  .option("password", "password")
  .option("driver", "com.mysql.jdbc.Driver")
  .load()

// 或者使用简写方式
val jdbcDF = spark.read
  .jdbc("jdbc:mysql://localhost:3306/mydb", "sales", 
        new java.util.Properties() {{
          setProperty("user", "username")
          setProperty("password", "password")
          setProperty("driver", "com.mysql.jdbc.Driver")
        }})

高级读取选项

// 分区读取（提高并行度）
val partitionedDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/mydb")
  .option("dbtable", "sales")
  .option("user", "username")
  .option("password", "password")
  .option("partitionColumn", "sale_id")        // 分区列
  .option("lowerBound", "1")                   // 最小值
  .option("upperBound", "1000000")             // 最大值
  .option("numPartitions", "10")               // 分区数
  .option("fetchsize", "1000")                 // 每次获取行数
  .load()

// 自定义查询读取
val customQueryDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/mydb")
  .option("query", """
    SELECT customer_id, SUM(amount) as total 
    FROM sales 
    WHERE sale_date >= '2024-01-01'
    GROUP BY customer_id
  """)
  .option("user", "username")
  .option("password", "password")
  .load()

SQL方式读取

-- 创建临时JDBC表
CREATE TEMPORARY VIEW jdbc_sales
USING org.apache.spark.sql.jdbc
OPTIONS (
  url "jdbc:mysql://localhost:3306/mydb",
  dbtable "sales",
  user "username",
  password "password",
  driver "com.mysql.jdbc.Driver"
);

-- 直接查询
SELECT * FROM jdbc_sales WHERE amount > 1000;

写入JDBC数据源

基本写入操作

// 写入JDBC表
resultDF.write
  .format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/mydb")
  .option("dbtable", "results")
  .option("user", "username")
  .option("password", "password")
  .option("driver", "com.mysql.jdbc.Driver")
  .mode("overwrite")  // 或 "append", "ignore", "error"
  .save()

// 批量写入优化
resultDF.write
  .format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/mydb")
  .option("dbtable", "results")
  .option("user", "username")
  .option("password", "password")
  .option("batchsize", "10000")        // 批量大小
  .option("isolationLevel", "READ_COMMITTED")
  .mode("append")
  .save()

事务性写入

// 确保数据一致性
import org.apache.spark.sql.SaveMode

try {
  resultDF.write
    .format("jdbc")
    .option("url", "jdbc:mysql://localhost:3306/mydb")
    .option("dbtable", "results")
    .option("user", "username")
    .option("password", "password")
    .option("batchsize", "10000")
    .mode(SaveMode.Append)
    .save()
} catch {
  case e: Exception =>
    println(s"写入失败: ${e.getMessage}")
    // 回滚或重试逻辑
}

HDFS 数据源集成

读取HDFS文件

不同文件格式读取

// Parquet格式（推荐）
val parquetDF = spark.read.parquet("hdfs://namenode:9000/data/sales.parquet")
// 或
val parquetDF = spark.read.format("parquet").load("hdfs://namenode:9000/data/sales.parquet")

// CSV格式
val csvDF = spark.read
  .format("csv")
  .option("header", "true")
  .option("inferSchema", "true")
  .option("delimiter", ",")
  .load("hdfs://namenode:9000/data/sales.csv")

// JSON格式
val jsonDF = spark.read.json("hdfs://namenode:9000/data/sales.json")

// ORC格式
val orcDF = spark.read.orc("hdfs://namenode:9000/data/sales.orc")

// 文本文件
val textDF = spark.read.text("hdfs://namenode:9000/data/logs.txt")

分区数据读取

// 自动发现分区
val partitionedDF = spark.read
  .parquet("hdfs://namenode:9000/data/sales/year=2024/month=*/day=*")

// 指定分区筛选
val januaryDF = spark.read
  .parquet("hdfs://namenode:9000/data/sales")
  .filter($"year" === 2024 && $"month" === 1)

写入HDFS文件

不同格式写入

// Parquet格式写入
resultDF.write
  .parquet("hdfs://namenode:9000/output/results.parquet")

// 带分区的写入
resultDF.write
  .partitionBy("year", "month")
  .parquet("hdfs://namenode:9000/output/partitioned_results")

// CSV格式写入
resultDF.write
  .format("csv")
  .option("header", "true")
  .option("delimiter", ",")
  .mode("overwrite")
  .save("hdfs://namenode:9000/output/results.csv")

// 压缩优化
resultDF.write
  .option("compression", "snappy")  // 或 "gzip", "lz4"
  .parquet("hdfs://namenode:9000/output/compressed_results")

写入模式控制

import org.apache.spark.sql.SaveMode

// 覆盖写入（默认）
resultDF.write.mode(SaveMode.Overwrite).parquet("hdfs://path")

// 追加写入
resultDF.write.mode(SaveMode.Append).parquet("hdfs://path")

// 忽略已存在
resultDF.write.mode(SaveMode.Ignore).parquet("hdfs://path")

// 错误模式（已存在则报错）
resultDF.write.mode(SaveMode.ErrorIfExists).parquet("hdfs://path")

其他数据源集成

Apache Hive 集成

// 启用Hive支持
val spark = SparkSession.builder()
  .appName("HiveIntegration")
  .config("spark.sql.warehouse.dir", "/user/hive/warehouse")
  .enableHiveSupport()
  .getOrCreate()

// 读取Hive表
val hiveDF = spark.sql("SELECT * FROM mydb.sales_table")

// 写入Hive表
resultDF.write.saveAsTable("mydb.result_table")

Amazon S3 集成

// S3配置
spark.sparkContext.hadoopConfiguration.set("fs.s3a.access.key", "your-access-key")
spark.sparkContext.hadoopConfiguration.set("fs.s3a.secret.key", "your-secret-key")
spark.sparkContext.hadoopConfiguration.set("fs.s3a.endpoint", "s3.amazonaws.com")

// 读写S3数据
val s3DF = spark.read.parquet("s3a://my-bucket/data/input.parquet")
s3DF.write.parquet("s3a://my-bucket/data/output.parquet")

Apache Kafka 集成

// 读取Kafka流
val kafkaDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load()

// 写入Kafka
resultDF.writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "output-topic")
  .start()

数据导入导出最佳实践

批量数据迁移示例

// 从JDBC到HDFS的完整ETL流程
def migrateJdbcToHdfs(): Unit = {
  // 1. 读取JDBC数据
  val jdbcDF = spark.read
    .format("jdbc")
    .option("url", "jdbc:mysql://localhost:3306/source_db")
    .option("dbtable", "source_table")
    .option("user", "user")
    .option("password", "pass")
    .option("partitionColumn", "id")
    .option("lowerBound", "1")
    .option("upperBound", "1000000")
    .option("numPartitions", "10")
    .load()
  
  // 2. 数据转换
  val transformedDF = jdbcDF
    .filter($"status" === "active")
    .withColumn("processed_date", current_date())
    .select($"id", $"name", $"amount", $"processed_date")
  
  // 3. 写入HDFS（分区+压缩）
  transformedDF.write
    .partitionBy("processed_date")
    .option("compression", "snappy")
    .mode("overwrite")
    .parquet("hdfs://namenode:9000/data/target/")
  
  println("数据迁移完成")
}

增量数据同步

// 增量同步策略
def incrementalSync(lastSyncTime: String): Unit = {
  // 读取增量数据
  val incrementalDF = spark.read
    .format("jdbc")
    .option("url", "jdbc:mysql://localhost:3306/source_db")
    .option("dbtable", s"(SELECT * FROM source_table WHERE update_time > '$lastSyncTime') as incremental")
    .option("user", "user")
    .option("password", "pass")
    .load()
  
  if (incrementalDF.count() > 0) {
    // 写入增量数据
    incrementalDF.write
      .mode("append")
      .parquet("hdfs://namenode:9000/data/incremental/")
    
    println(s"增量同步完成，处理了${incrementalDF.count()}条记录")
  } else {
    println("没有新的增量数据")
  }
}

性能优化配置

连接池和并行度优化

// JDBC性能优化配置
val optimizedJdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/mydb")
  .option("dbtable", "large_table")
  .option("user", "user")
  .option("password", "pass")
  .option("partitionColumn", "id")
  .option("lowerBound", "1")
  .option("upperBound", "10000000")
  .option("numPartitions", "50")                // 根据集群规模调整
  .option("fetchsize", "5000")                  // 增大fetch大小
  .option("queryTimeout", "3600")               // 超时时间
  .option("sessionInitStatement", "SET SESSION group_concat_max_len = 1000000")
  .load()

文件格式选择建议

// 不同场景的文件格式选择
def chooseFileFormat(scenario: String): Unit = scenario match {
  case "analytics" => 
    // 分析查询：Parquet（列式存储，压缩率高）
    df.write.parquet("hdfs://path/analytics_data")
  
  case "streaming" =>
    // 流式处理：ORC（更好的ACID支持）
    df.write.orc("hdfs://path/streaming_data")
  
  case "interchange" =>
    // 数据交换：CSV/JSON（通用性好）
    df.write.option("header", "true").csv("hdfs://path/interchange_data")
  
  case _ =>
    df.write.parquet("hdfs://path/default_data")
}

错误处理和监控

健壮的数据导入导出

import scala.util.{Try, Success, Failure}

def safeDataTransfer(sourcePath: String, targetPath: String): Boolean = {
  Try {
    // 读取源数据
    val sourceDF = spark.read.parquet(sourcePath)
    
    // 数据验证
    require(sourceDF.count() > 0, "源数据为空")
    
    // 写入目标
    sourceDF.write
      .mode("overwrite")
      .option("compression", "snappy")
      .parquet(targetPath)
      
    // 验证写入结果
    val writtenDF = spark.read.parquet(targetPath)
    require(writtenDF.count() == sourceDF.count(), "数据量不匹配")
    
    true
  } match {
    case Success(_) => 
      println("数据转移成功")
      true
    case Failure(e) =>
      println(s"数据转移失败: ${e.getMessage}")
      false
  }
}

Spark SQL通过统一的数据源API提供了强大的外部数据集成能力，支持多种数据格式和存储系统，使得数据导入导出变得简单高效。