如何使用Spark读取Oracle数据库中的所有数据表？

Hey there! Great question—Spark doesn’t have a built-in equivalent to Sqoop’s import-all-tables out of the box, but we can absolutely replicate that functionality, and even fix the repeated connection issue you’re facing. Here’s how to do it properly:

Step 1: Fetch All Table Names from Oracle #

First, we need to get a list of all tables you want to read. We can do this with a quick JDBC call to Oracle’s system views (use USER_TABLES for tables owned by your user, or ALL_TABLES if you have access to other schemas). This only requires one connection upfront, not per table.

import java.sql.DriverManager

// Configure your Oracle connection details
val oracleUrl = "jdbc:oracle:thin:@//your-host:port/your-service-name"
val oracleUser = "your-username"
val oraclePass = "your-password"

// Establish a single connection to fetch table names
val conn = DriverManager.getConnection(oracleUrl, oracleUser, oraclePass)
val stmt = conn.createStatement()

// Adjust the query based on your needs (e.g., filter out system tables)
val tableNameQuery = "SELECT TABLE_NAME FROM USER_TABLES WHERE TABLE_NAME NOT LIKE 'SYS_%'"
val rs = stmt.executeQuery(tableNameQuery)

// Collect table names into a list
val tableNames = scala.collection.mutable.ListBuffer[String]()
while (rs.next()) {
  tableNames += rs.getString("TABLE_NAME")
}

// Clean up resources
rs.close()
stmt.close()
conn.close()

Step 2: Batch Read Tables with Reused Connections #

Now that we have our table list, we can read each table in a loop. The key here is to use Spark’s built-in connection pooling to avoid creating a new connection for every table. Spark 2.3+ uses HikariCP by default for JDBC connections, so it will reuse connections automatically if we use a consistent configuration.

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("OracleFullImport")
  // Optional: Explicitly set connection pool (default is HikariCP)
  .config("spark.sql.jdbc.connectionPool", "HikariCP")
  .getOrCreate()

// Define reusable JDBC configuration
val jdbcConfig = Map(
  "url" -> oracleUrl,
  "user" -> oracleUser,
  "password" -> oraclePass,
  "driver" -> "oracle.jdbc.driver.OracleDriver"
)

// Read each table and process/store as needed
tableNames.foreach { tableName =>
  println(s"Reading table: $tableName")
  val tableDF = spark.read.jdbc(
    url = jdbcConfig("url"),
    table = tableName,
    properties = jdbcConfig.filter(_._1 != "url").asJava
  )

  // Example: Save table to Parquet (adjust output format as needed)
  tableDF.write.mode("overwrite")
    .parquet(s"/your/output/path/oracle_tables/$tableName")
}

Bonus: Optimizations & Extras #

• Parallelize Large Tables: For big tables, add partitioning parameters (numPartitions, partitionColumn, lowerBound, upperBound) to the JDBC read to speed up ingestion.
• Union All Tables: If you want to combine all tables into a single DataFrame (with a column to track the source table), use this snippet:

  import org.apache.spark.sql.functions.lit
  
  val combinedDF = tableNames.foldLeft(spark.emptyDataFrame) { (accDF, tableName) =>
    val currentDF = spark.read.jdbc(jdbcConfig("url"), tableName, jdbcConfig.filter(_._1 != "url").asJava)
      .withColumn("source_table", lit(tableName))
    accDF.unionByName(currentDF, allowMissingColumns = true)
  }
  

• Filter Tables: Modify the initial tableNameQuery to exclude unwanted tables (e.g., system tables, temporary tables) to avoid unnecessary work.

This approach not only mimics Sqoop’s import-all-tables but also fixes your repeated connection issue by leveraging Spark’s connection pooling.

内容的提问来源于stack exchange，提问作者Surajit Kundu