> ## Documentation Index
> Fetch the complete documentation index at: https://cockroachlabs.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Migrate Data with Debezium

export const InternalLink = ({version, path = "", children, ...props}) => {
  let detectedVersion = version || "stable";
  if (typeof window !== 'undefined' && !version) {
    const match = window.location.pathname.match(/\/docs\/([^/]+)/);
    if (match) {
      detectedVersion = match[1];
    }
  }
  const normalizedPath = path.startsWith("/") ? path.slice(1) : path;
  return <a href={`/docs/${detectedVersion}/${normalizedPath}`} {...props}>
      {children}
    </a>;
};

[Debezium](https://debezium.io/) is a self-hosted distributed platform that can read data from a variety of sources and import it into Kafka. You can use Debezium to [migrate data to CockroachDB](#migrate-data-to-cockroachdb) from another database that is accessible over the public internet.

As of this writing, Debezium supports the following database [sources](https://debezium.io/documentation/reference/stable/connectors/index.html):

* MongoDB
* MySQL
* PostgreSQL
* SQL Server
* Oracle
* Db2
* Cassandra
* Vitess (incubating)
* Spanner (incubating)
* JDBC (incubating)

<Note>
  Migrating with Debezium requires familiarity with Kafka. Refer to the [Debezium documentation](https://debezium.io/documentation/reference/stable/architecture.html) for information on how Debezium is deployed with Kafka Connect.
</Note>

## Before you begin

Complete the following items before using Debezium:

* Configure a secure <InternalLink version="cockroachcloud" path="network-authorization">publicly-accessible</InternalLink> CockroachDB cluster running the latest <InternalLink version="releases" path="v25.1">production release</InternalLink> with at least one <InternalLink path="security-reference/authorization#sql-users">SQL user</InternalLink>, make a note of the credentials for the SQL user.
* Install and configure [Debezium](https://debezium.io/), [Kafka Connect](https://docs.confluent.io/platform/current/connect/index.html), and [Kafka](https://kafka.apache.org/).

## Migrate data to CockroachDB

Once all of the [prerequisite steps](#before-you-begin) are completed, you can use Debezium to migrate data to CockroachDB.

1. To write data from Kafka to CockroachDB, use the Confluent JDBC Sink Connector. First use the following `dockerfile` to create a custom image with the [JDBC driver](https://www.confluent.io/hub/confluentinc/kafka-connect-jdbc):

   ```
   FROM quay.io/debezium/connect:latest
   ENV KAFKA_CONNECT_JDBC_DIR=$KAFKA_CONNECT_PLUGINS_DIR/kafka-connect-jdbc

   ARG POSTGRES_VERSION=latest
   ARG KAFKA_JDBC_VERSION=latest

   # Deploy PostgreSQL JDBC Driver
   RUN cd /kafka/libs && curl -sO https://jdbc.postgresql.org/download/postgresql-$POSTGRES_VERSION.jar

   # Deploy Kafka Connect JDBC
   RUN mkdir $KAFKA_CONNECT_JDBC_DIR && cd $KAFKA_CONNECT_JDBC_DIR &&\
      curl -sO https://packages.confluent.io/maven/io/confluent/kafka-connect-jdbc/$KAFKA_JDBC_VERSION/kafka-connect-jdbc-$KAFKA_JDBC_VERSION.jar
   ```

2. Create the JSON configuration file that you will use to add data from your [source database](https://debezium.io/documentation/reference/stable/connectors/index.html) to a Kafka topic. For example:

   ```shell theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}}
   {
     "name": "pg-source",
     "config": {
       "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
       "database.dbname": "{database}",
       "database.hostname": "{hostname}",
       "database.password": "",
       "database.port": "5432",
       "database.user": "postgres",
       "plugin.name": "pgoutput",
       "table.include.list": "public.test_table_small",
       "tasks.max": "1",
       "topic.creation.default.cleanup.policy": "delete",
       "topic.creation.default.partitions": "10",
       "topic.creation.default.replication.factor": "1",
       "topic.creation.default.retention.ms": "604800000",
       "topic.creation.enable": "true",
       "topic.prefix": "{username}",
       "slot.name" : "debezium"
     }
   }
   ```

3. Create the JSON configuration file that you will use to create the sink. For example:

   ```shell theme={"theme":{"light":"catppuccin-mocha","dark":"catppuccin-mocha"}}
   {
     "name": "pg-sink",
     "config": {
       "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
       "tasks.max": "10",
       "topics" : "{topic.example.table}",
       "connection.url": "jdbc:postgresql://{host}:{port}/{username}?sslmode=require",
       "connection.user": "{username}",
       "connection.password": "{password}",
       "insert.mode": "upsert",
       "pk.mode": "record_value",
       "pk.fields": "id",
       "database.time_zone": "UTC",
       "auto.create": true,
       "auto.evolve": false,
       "transforms": "unwrap",
       "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState"
     }
   }
   ```

   Specify `connection.url` in <InternalLink path="connect-to-the-database?filters=java&">JDBC format</InternalLink>. For information about where to find the CockroachDB connection parameters, see <InternalLink path="connect-to-the-database">Connect to a CockroachDB Cluster</InternalLink>.

   The preceding snippet is an example configuration. For details on the configurable fields, see the [Confluent JDBC Sink Connector documentation](https://docs.confluent.io/kafka-connectors/jdbc/current/sink-connector/sink_config_options.html).

4. To create the sink, `POST` the JSON configuration file to the Kafka Connect `/connectors` endpoint. Refer to the [Kafka Connect API documentation](https://kafka.apache.org/documentation/#connect_rest) for more information.

## See also

* <InternalLink version="molt" path="migration-overview">Migration Overview</InternalLink>
* <InternalLink version="cockroachcloud" path="migrations-page">Schema Conversion Tool</InternalLink>
* <InternalLink path="change-data-capture-overview">Change Data Capture Overview</InternalLink>
* <InternalLink path="third-party-database-tools">Third-Party Tools Supported by Cockroach Labs</InternalLink>
* <InternalLink path="stream-a-changefeed-to-a-confluent-cloud-kafka-cluster">Stream a Changefeed to a Confluent Cloud Kafka Cluster</InternalLink>