无需Kafka，如何将Oracle表作为Spark Streaming实时数据源？

Yes, you absolutely can! Since Kafka is off-limits, you’ll rely on Oracle’s Change Data Capture (CDC) capabilities paired with Spark’s dedicated connectors to stream real-time insert events directly from Oracle. Here are the most practical approaches:

1. Use the Oracle CDC Connector for Spark Structured Streaming #

This is the most straightforward and officially supported way. Oracle provides a connector that integrates directly with Spark Structured Streaming, letting you capture row-level changes (including inserts) from Oracle’s redo logs or archive logs.

Key steps to set this up: #

• Enable CDC on your Oracle table: First, enable supplemental logging for the target table (or entire database) to ensure Oracle tracks all necessary change details. Run these SQL commands:

  ALTER TABLE your_target_table ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
  ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
  

• Add the Oracle CDC Connector dependency: Include the connector JAR in your Spark job (via the --jars flag or build tools like Maven/Gradle).
• Configure your Spark Structured Streaming query: Write a streaming DataFrame that connects to Oracle using the CDC source. Sample code snippet:

  val df = spark.readStream
    .format("oracle.cdc")
    .option("db.url", "jdbc:oracle:thin:@//your-oracle-host:1521/your-db-service")
    .option("db.user", "your-username")
    .option("db.password", "your-password")
    .option("table.name", "your_schema.your_target_table")
    .option("start.position", "latest") // Start capturing from the newest change
    .load()
  
  // Process and output streaming data (example writes to console)
  val query = df.writeStream
    .outputMode("append")
    .format("console")
    .start()
  
  query.awaitTermination()
  

2. Polling with a Tracking Column (Fallback Option) #

If you can’t use the CDC connector (e.g., due to restricted access to redo logs), you can implement a polling mechanism using a timestamp or sequence column in your Oracle table. This is simpler but less efficient than CDC.

How it works: #

• Add a tracking column: Ensure your table has a created_timestamp (or auto-incrementing sequence_id) that populates automatically when a row is inserted.
• Use Spark’s JDBC streaming source: Set up a query that periodically polls for new rows based on the tracking column, using a watermark to avoid reprocessing old data. Example code:

  val df = spark.readStream
    .format("jdbc")
    .option("url", "jdbc:oracle:thin:@//your-oracle-host:1521/your-db-service")
    .option("dbtable", "(SELECT * FROM your_schema.your_target_table WHERE created_timestamp > ?) AS latest_rows")
    .option("user", "your-username")
    .option("password", "your-password")
    .option("fetchsize", "1000")
    .option("spark.sql.streaming.schemaInference", "true")
    .load()
  
  // Add watermark to handle late-arriving data
  val processedDf = df.withWatermark("created_timestamp", "10 minutes")
    .groupBy(window($"created_timestamp", "5 minutes"))
    .count()
  
  val query = processedDf.writeStream
    .outputMode("append")
    .format("console")
    .start()
  
  query.awaitTermination()
  

  Note: This approach has latency tied to your polling interval, unlike CDC which captures changes near-real-time.

Important Considerations #

• Permissions: For the CDC connector, your Oracle user needs access to redo logs and permission to use Oracle LogMiner (the underlying mechanism for some CDC implementations).
• Latency: CDC offers near-real-time capture (sub-second to a few seconds), while polling latency equals your configured interval.
• Data Consistency: Double-check supplemental logging is correctly set up to avoid missing change events.
• Compatibility: Ensure the Oracle CDC connector version matches your Spark version (e.g., Spark 3.3+ works with the latest connector releases).

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