无主键Oracle表增量导入HDFS的Sqoop实现及其他方案咨询

Hey there, let's break down how to tackle your incremental data import from Oracle to HDFS—especially with those tricky tables without primary keys or with composite ones. Here's a practical guide using Sqoop first, followed by alternative tools you can consider:

Sqoop is the go-to tool for batch data transfer between relational databases and Hadoop, but we need to adapt its approach based on your table structure.

1. Tables with Single Primary Keys #

This is the straightforward case. Sqoop's --incremental flag works seamlessly here, with two modes:

• Append mode: For tables where new rows are added and never updated (use a numeric primary key as the check column).
  Example command:

  sqoop import \
    --connect jdbc:oracle:thin:@//your-oracle-host:1521/your-db \
    --username your-username \
    --password your-password \
    --table your_table \
    --incremental append \
    --check-column id \
    --last-value 1000 \  # Replace with the max ID from your last import
    --target-dir /hdfs/path/to/your_table \
    --append
  

• Lastmodified mode: For tables with timestamp columns tracking when rows were created/updated. Use this if you need to capture updates too (pair with --merge-key to overwrite old rows in HDFS).
  Example command:

  sqoop import \
    --connect jdbc:oracle:thin:@//your-oracle-host:1521/your-db \
    --username your-username \
    --password your-password \
    --table your_table \
    --incremental lastmodified \
    --check-column update_time \
    --last-value "2024-01-01 00:00:00" \  # Replace with the latest timestamp from last import
    --target-dir /hdfs/path/to/your_table \
    --merge-key id  # Merge updates using the primary key
  

2. Tables with Composite Primary Keys #

Sqoop doesn't natively support composite keys as check columns, but we can work around this with two approaches:

• Option 1: Create a virtual composite key in Oracle
  Concatenate your composite key columns into a single string (e.g., col1 || '-' || col2) and use that as a pseudo-check column. Use --query instead of --table to include this virtual column:

  sqoop import \
    --connect jdbc:oracle:thin:@//your-oracle-host:1521/your-db \
    --username your-username \
    --password your-password \
    --query "SELECT *, col1 || '-' || col2 AS composite_key FROM your_table WHERE \$CONDITIONS" \
    --incremental append \
    --check-column composite_key \
    --last-value "100-200" \  # Replace with the last composite key value from your last import
    --target-dir /hdfs/path/to/your_table \
    --split-by col1 \  # Split data using one of the composite key columns for parallelism
    --append
  

• Option 2: Build a custom WHERE clause
  Track the last imported values of each composite key column, then write a query that filters for rows beyond that combination. For example, if your composite key is (col1, col2) and the last imported values were col1=100, col2=200, your query would be:

  SELECT * FROM your_table WHERE col1 > 100 OR (col1 = 100 AND col2 > 200)
  

  Automate this with a shell script that stores the last key values and injects them into the Sqoop command.

3. Tables Without Primary Keys #

This is the most challenging scenario—we need to find alternative ways to identify incremental data:

• Use a business-unique incremental column: If your table has a timestamp (e.g., create_time) or auto-incrementing sequence column, treat it like a pseudo-primary key and use the same --incremental approach as single-key tables.
• Full import + delta comparison: Import the entire table to a temporary HDFS directory, then use Hive or Spark to compare it with your existing data and only keep new/updated rows. This works if your tables aren't too large to full-import regularly.
• Enable Oracle CDC: Use Oracle's built-in Change Data Capture (CDC) or Oracle GoldenGate to track row-level changes (inserts/updates/deletes). Sqoop can then import the CDC log tables, which include metadata about changes—no need for primary keys.
• Custom incremental query: If you have a column that can filter new rows (like create_time), use --query to fetch only rows added since your last import:

  sqoop import \
    --connect jdbc:oracle:thin:@//your-oracle-host:1521/your-db \
    --username your-username \
    --password your-password \
    --query "SELECT * FROM your_table WHERE create_time > TO_DATE('2024-01-01 00:00:00', 'YYYY-MM-DD HH24:MI:SS') AND \$CONDITIONS" \
    --target-dir /hdfs/path/to/your_table \
    --split-by some_column \  # Pick a column with even data distribution for parallelism
    --append
  

  Make sure to track the create_time threshold in a file or database to automate future runs.

If Sqoop's limitations feel too restrictive, these tools offer more flexibility:

1. Apache Spark #

Spark's JDBC connector lets you write custom incremental logic with full control over data processing. It's great for complex scenarios (e.g., multi-table joins, data cleaning) and handles parallelism efficiently.
Example Scala code snippet:

import org.apache.spark.sql.SparkSession
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

val spark = SparkSession.builder().appName("OracleIncrementalImport").getOrCreate()

// Read the last import timestamp from HDFS
val lastImportTime = spark.read.textFile("/hdfs/path/last_import_time").first()

// Fetch incremental data from Oracle
val incrementalDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:oracle:thin:@//your-oracle-host:1521/your-db")
  .option("dbtable", s"(SELECT * FROM your_table WHERE create_time > TO_DATE('$lastImportTime', 'YYYY-MM-DD HH24:MI:SS')) t")
  .option("user", "your-username")
  .option("password", "your-password")
  .load()

// Append to HDFS
incrementalDF.write.mode("append").parquet("/hdfs/path/to/your_table")

// Update the last import timestamp
val currentTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
spark.sparkContext.parallelize(Seq(currentTime)).saveAsTextFile("/hdfs/path/last_import_time")

2. Apache Flume (JDBC Source) #

Flume is ideal for near-real-time incremental imports. Configure its JDBC source to poll Oracle at intervals and write directly to HDFS.
Sample flume.conf snippet:

agent.sources = jdbcSource
agent.sinks = hdfsSink
agent.channels = memoryChannel

# JDBC Source Configuration
agent.sources.jdbcSource.type = org.apache.flume.source.jdbc.JDBCSource
agent.sources.jdbcSource.jdbc.driver = oracle.jdbc.OracleDriver
agent.sources.jdbcSource.jdbc.url = jdbc:oracle:thin:@//your-oracle-host:1521/your-db
agent.sources.jdbcSource.jdbc.user = your-username
agent.sources.jdbcSource.jdbc.password = your-password
agent.sources.jdbcSource.query = SELECT * FROM your_table WHERE create_time > ?
agent.sources.jdbcSource.incremental.column.name = create_time
agent.sources.jdbcSource.incremental.column.type = TIMESTAMP
agent.sources.jdbcSource.run.delay = 3600000  # Poll every hour (adjust as needed)

# HDFS Sink Configuration
agent.sinks.hdfsSink.type = hdfs
agent.sinks.hdfsSink.hdfs.path = hdfs://namenode:9000/path/to/your_table
agent.sinks.hdfsSink.hdfs.filePrefix = your_table_
agent.sinks.hdfsSink.hdfs.fileSuffix = .parquet
agent.sinks.hdfsSink.hdfs.writeFormat = Parquet
agent.sinks.hdfsSink.hdfs.fileType = DataStream

# Channel Configuration
agent.channels.memoryChannel.type = memory
agent.channels.memoryChannel.capacity = 10000
agent.channels.memoryChannel.transactionCapacity = 1000

# Bind Components
agent.sources.jdbcSource.channels = memoryChannel
agent.sinks.hdfsSink.channel = memoryChannel

3. Oracle GoldenGate #

For enterprise-grade real-time synchronization, Oracle GoldenGate captures changes directly from Oracle's redo logs, so you don't rely on primary keys or timestamp columns. It can stream incremental changes directly to HDFS (via GoldenGate for Big Data) and supports inserts, updates, and deletes.

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