无需第三方平台，如何将Kafka数据导出至Oracle数据库？

Hey there! Great question—you absolutely don’t need Confluent Platform to move Kafka data into Oracle. Here’s a straightforward breakdown of what you’ll need and how to make it work:

Core Tools & Components #

You can stick purely to Apache Kafka’s native ecosystem and open-source plugins:

• Apache Kafka Cluster: A self-hosted (or even local) Apache Kafka deployment (not Confluent’s managed platform). This includes the Kafka brokers and the Kafka Connect component—Connect is built into Apache Kafka, so you don’t need to install anything extra for it.
• JDBC Sink Connector: While this is often associated with Confluent, it’s an open-source, Apache 2.0-licensed plugin that you can download independently. You don’t need the full Confluent Platform to use it.
• Oracle JDBC Driver: The official ojdbc.jar file (matching your Oracle database version) to enable connectivity between Connect and Oracle.

Prerequisites #

Before you start, make sure you have these covered:

• A running Apache Kafka cluster with at least one broker, and Kafka Connect configured (either standalone or distributed mode—distributed is better for production).
• An Oracle database instance with a user account that has permissions to: create tables, insert/update data, and alter tables (if you want auto-schema evolution).
• The Oracle JDBC driver downloaded and placed in Kafka Connect’s classpath (e.g., $KAFKA_HOME/libs or a directory specified by the plugin.path setting in your Connect config).
• The JDBC Sink Connector JAR file added to the same classpath/plugin directory as the Oracle driver.

Key Configuration Steps #

1. Set Up Kafka Connect Worker
   First, configure your Connect worker (edit the connect-distributed.properties or connect-standalone.properties file):

   • Set bootstrap.servers to point to your Kafka brokers.
   • Choose a converter for message keys/values. For simplicity, use the native JsonConverter (no need for Schema Registry):

     key.converter=org.apache.kafka.connect.json.JsonConverter
     value.converter=org.apache.kafka.connect.json.JsonConverter
     value.converter.schemas.enable=true
     

   • Specify plugin.path to the directory where you placed the JDBC Sink Connector and Oracle driver JARs.

2. Configure the JDBC Sink Connector
   Create a JSON config file for the sink connector (example below) and submit it to the Connect REST API:

   {
     "name": "oracle-kafka-sink",
     "config": {
       "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
       "tasks.max": "3",
       "topics": "your_source_kafka_topic",
       "connection.url": "jdbc:oracle:thin:@//your_oracle_host:1521/your_service_name",
       "connection.user": "oracle_username",
       "connection.password": "oracle_password",
       "auto.create": "true",
       "auto.evolve": "true",
       "insert.mode": "upsert",
       "pk.fields": "your_primary_key_column",
       "value.converter": "org.apache.kafka.connect.json.JsonConverter",
       "value.converter.schemas.enable": "true"
     }
   }
   

   Let’s break down the critical settings:

   • connection.url: Follow Oracle’s JDBC thin driver format (adjust host, port, and service name to match your setup).
   • auto.create: Lets the connector automatically create Oracle tables matching your Kafka message schema.
   • auto.evolve: Updates table schemas if your Kafka message structure changes (use cautiously in production).
   • insert.mode: Choose insert for pure appends, or upsert to handle updates based on a primary key.

3. Start the Connector
   Submit the config to Connect using the REST API:

   curl -X POST -H "Content-Type: application/json" --data @sink-config.json http://your-connect-host:8083/connectors
   

Quick Tips #

• If your Kafka messages are raw strings (not structured JSON), you’ll need to use a custom converter or pre-process the data to map it to Oracle table columns.
• For production, scale the tasks.max value to match your throughput needs, and tune batch.size to optimize bulk inserts into Oracle.
• Always test the setup with a small dataset first to validate schema mapping and connectivity.

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