<commit_before>---
layout: post
title:  Streaming data to a downstream database
date:   2017-09-25 10:19:59 -0600
tags: [ mysql, postgres, smt, example ]
author: jpechane
---

In this blog post we will create a simple streaming data pipeline to continuously capture the changes in a MySQL database and replicate them in near real-time into a PostgreSQL database.
We'll show how to do this without writing any code, but instead by using and configuring Kafka Connect, the Debezium MySQL source connector, the Confluent JDBC sink connector, and a few single message transforms (SMTs).

This approach of replicating data through Kafka is really useful on its own, but it becomes even more advantageous when we can combine our near real-time streams of data changes with other streams, connectors, and stream processing applications.
A recent https://www.confluent.io/blog/simplest-useful-kafka-connect-data-pipeline-world-thereabouts-part-1/[Confluent blog post series] shows a similar streaming data pipeline but using different connectors and SMTs.
What's great about Kafka Connect is that you can mix and match connectors to move data between multiple systems.

We will also demonstrate a new functionality that was released with link:2017/09/21/debezium-0-6-0-released/[Debezium 0.6.0]: a single message transform for link:/docs/configuration/event-flattening/[CDC Event Flattening].

+++<!-- more -->+++

== Topology
The general topology for this scenario is displayed on the following picture:

[.centered-image.responsive-image]
====
++++
<img src="/assets/images/dbz-to-jdbc.svg" style="max-width:100%;" class="responsive-image">
++++
Figure 1: A General topology
====

&nbsp; +

To simplify the setup a little bit, we will use only one Kafka Connect instance that will contain all connectors.
I.e. this instance will serve as an event producer and an event consumer:

&nbsp; +

[.centered-image.responsive-image]
====
++++
<img src="/assets/images/dbz-to-jdbc-simplified.svg" style="max-width:100%;" class="responsive-image">
++++
Figure 2: A Simplified topology
====

== Configuration
We will use this https://github.com/debezium/debezium-examples/tree/main/unwrap-smt[compose] for a fast deployment of the demo.
The deployment consists of following Docker images:

* https://hub.docker.com/r/debezium/zookeeper/[Apache ZooKeeper]
* https://hub.docker.com/r/debezium/kafka/[Apache Kafka]
* An https://github.com/debezium/debezium-examples/tree/main/unwrap-smt/debezium-jdbc[enriched] Kafka Connect / Debezium https://hub.docker.com/r/debezium/connect/[image] with changes
** PostgreSQL JDBC driver placed into `/kafka/libs` directory
** https://docs.confluent.io/current/connect/connect-jdbc/docs/index.html[Kafka Connect JDBC Connector] (developed by https://www.confluent.io/[Confluent]) placed into `/kafka/connect/kafka-connect-jdbc` directory
* Pre-populated MySQL used in our link:docs/tutorial[tutorial]
* Empty PostgreSQL

The Debezium MySQL Connector was designed to specifically capture database changes and provide as much information as possible about those events beyond just the new state of each row.
Meanwhile, the Confluent JDBC Sink Connector was designed to simply convert each message into a database insert/upsert based upon the structure of the message.
So, the two connectors have different structures for the messages, but they also use different topic naming conventions and behavior of representing deleted records.

These mismatches in structure and behavior will be common when using connectors that were not designed to work together. But this is something that we can easily deal with, and we discuss how in the next few sections.

=== Event format
Debezium emits events in a complex format that contains all of the information about the captured data change:
the type of operation, source metadata, the timestamp the event was processed by the connector, and state of the row before and after the change was made.
Debezium calls this structure an _"envelope"_:

[source,json,indent=0]
----
{
	"op": "u",
	"source": {
		...
	},
	"ts_ms" : "...",
	"before" : {
		"field1" : "oldvalue1",
		"field2" : "oldvalue2"
	},
	"after" : {
		"field1" : "newvalue1",
		"field2" : "newvalue2"
	}
}
----

Many other Kafka Connect source connectors don't have the luxury of knowing this much about the changes, and instead use a simpler model where each message directly represents the after state of the row.
This is also what many sink connectors expect, and the Confluent JDBC Sink Connector is not different:

[source,json,indent=0]
----
{
	"field1" : "newvalue1",
	"field2" : "newvalue2"
}
----

While we think it's actually a great thing that Debezium CDC connectors provide as much detail as possible, we also make it easy for you to transform Debezium's _"envelope"_ format into the _"row"_ format that is expected by many other connectors.
Debezium provides a bridge between those two formats in a form of a https://cwiki.apache.org/confluence/display/KAFKA/KIP-66%3A+Single+Message+Transforms+for+Kafka+Connect[single message transform].
The `UnwrapFromEnvelope` transformation automatically extracts a new row record and thus effectively _flattens_ the complex record into a simple one consumable by other connectors.

You can use this SMT on the source connector to transform the message _before_ it is written to Kafka, or you can instead store the source connector's richer _"envelope"_ form of the message in Kafka and use this SMT on the sink connector to transform the message _after_ it is read from Kafka and before it is passed to the sink connector.
Both options work, and it just depends on whether you find the envelope form of the message useful for other purposes.

In our example we apply the SMT at the sink connector using these configuration properties:

[source,indent=0]
----
        "transforms": "unwrap",
        "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
----

=== Delete records
When the Debezium connector detects a row is deleted, it creates two event messages: a _delete_ event and a _tombstone_ message.
The _delete_ message has an envelope with the state of the deleted row in the `before` field, and an `after` field that is `null`.
The _tombstone_ message contains same key as the _delete_ message, but the entire message value is `null`, and Kafka's log compaction utilizes this to know that it can remove any earlier messages with the same key.
A number of sink connectors, including the Confluent's JDBC Sink Connector, are not expecting these messages and will instead fail if they see either kind of message.
The `UnwrapFromEnvelope` SMT will by default filter out both _delete_ and _tombstone_ records, though you can change this if you're using the SMT and want to keep one or both of these kinds of messages.

=== Topic naming
Last but not least there is a difference in naming of topics.
Debezium uses fully qualified naming for target topics representing each table it manages.
The naming follows the pattern `<logical-name>.<database-name>.<table-name>`.
Kafka Connect JDBC Connector works with simple names `<table-name>`.

In more complex scenarios the user may deploy the https://kafka.apache.org/documentation/streams/[Kafka Streams] framework to establish elaborated routing between source and target routes.
In our example we will use a stock `RegexRouter` SMT that would route records created by Debezium into topics named according to JDBC Connector schema.
Again, we could use this SMT in either the source or sink connectors, but for this example we're going to use it in the source connector so we can choose the names of the Kafka topics where the records will be written.

[source,indent=0]
----
        "transforms": "route",
        "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
        "transforms.route.replacement": "$3"
----

== Example
Kick the tires and let's try our example!

First of all we need to deploy all components.
[source,bash,indent=0]
----
export DEBEZIUM_VERSION=0.6
docker-compose up
----

When all components are started we are going to register the JDBC Sink connector writing into PostgreSQL database:
[source,bash,indent=0]
----
curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @jdbc-sink.json
----

Using this registration request:

[source,json,indent=0]
----
{
    "name": "jdbc-sink",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "topics": "customers",
        "connection.url": "jdbc:postgresql://postgres:5432/inventory?user=postgresuser&password=<PASSWORD>",
        "transforms": "unwrap",                                                  (1)
        "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",   (1)
        "auto.create": "true",                                                   (2)
        "insert.mode": "upsert",                                                 (3)
        "pk.fields": "id",                                                       (4)
        "pk.mode": "record_value"                                                (4)
    }
}
----

The request configures these options:

1. unwrapping Debezium's complex format into a simple one
2. automatically create target tables
3. insert a row if it does not exist or update an existing one
4. identify the primary key stored in Kafka's record value field

Then the source connector must be set up:

[source,bash,indent=0]
----
curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @source.json
----<commit_msg>Update obsolete transform name not to confuse people reading this blog post<commit_after>---
layout: post
title:  Streaming data to a downstream database
date:   2017-09-25 10:19:59 -0600
tags: [ mysql, postgres, smt, example ]
author: jpechane
---

In this blog post we will create a simple streaming data pipeline to continuously capture the changes in a MySQL database and replicate them in near real-time into a PostgreSQL database.
We'll show how to do this without writing any code, but instead by using and configuring Kafka Connect, the Debezium MySQL source connector, the Confluent JDBC sink connector, and a few single message transforms (SMTs).

This approach of replicating data through Kafka is really useful on its own, but it becomes even more advantageous when we can combine our near real-time streams of data changes with other streams, connectors, and stream processing applications.
A recent https://www.confluent.io/blog/simplest-useful-kafka-connect-data-pipeline-world-thereabouts-part-1/[Confluent blog post series] shows a similar streaming data pipeline but using different connectors and SMTs.
What's great about Kafka Connect is that you can mix and match connectors to move data between multiple systems.

We will also demonstrate a new functionality that was released with link:2017/09/21/debezium-0-6-0-released/[Debezium 0.6.0]: a single message transform for link:/docs/configuration/event-flattening/[CDC Event Flattening].

+++<!-- more -->+++

== Topology
The general topology for this scenario is displayed on the following picture:

[.centered-image.responsive-image]
====
++++
<img src="/assets/images/dbz-to-jdbc.svg" style="max-width:100%;" class="responsive-image">
++++
Figure 1: A General topology
====

&nbsp; +

To simplify the setup a little bit, we will use only one Kafka Connect instance that will contain all connectors.
I.e. this instance will serve as an event producer and an event consumer:

&nbsp; +

[.centered-image.responsive-image]
====
++++
<img src="/assets/images/dbz-to-jdbc-simplified.svg" style="max-width:100%;" class="responsive-image">
++++
Figure 2: A Simplified topology
====

== Configuration
We will use this https://github.com/debezium/debezium-examples/tree/main/unwrap-smt[compose] for a fast deployment of the demo.
The deployment consists of following Docker images:

* https://hub.docker.com/r/debezium/zookeeper/[Apache ZooKeeper]
* https://hub.docker.com/r/debezium/kafka/[Apache Kafka]
* An https://github.com/debezium/debezium-examples/tree/main/unwrap-smt/debezium-jdbc[enriched] Kafka Connect / Debezium https://hub.docker.com/r/debezium/connect/[image] with changes
** PostgreSQL JDBC driver placed into `/kafka/libs` directory
** https://docs.confluent.io/current/connect/connect-jdbc/docs/index.html[Kafka Connect JDBC Connector] (developed by https://www.confluent.io/[Confluent]) placed into `/kafka/connect/kafka-connect-jdbc` directory
* Pre-populated MySQL used in our link:docs/tutorial[tutorial]
* Empty PostgreSQL

The Debezium MySQL Connector was designed to specifically capture database changes and provide as much information as possible about those events beyond just the new state of each row.
Meanwhile, the Confluent JDBC Sink Connector was designed to simply convert each message into a database insert/upsert based upon the structure of the message.
So, the two connectors have different structures for the messages, but they also use different topic naming conventions and behavior of representing deleted records.

These mismatches in structure and behavior will be common when using connectors that were not designed to work together. But this is something that we can easily deal with, and we discuss how in the next few sections.

=== Event format
Debezium emits events in a complex format that contains all of the information about the captured data change:
the type of operation, source metadata, the timestamp the event was processed by the connector, and state of the row before and after the change was made.
Debezium calls this structure an _"envelope"_:

[source,json,indent=0]
----
{
	"op": "u",
	"source": {
		...
	},
	"ts_ms" : "...",
	"before" : {
		"field1" : "oldvalue1",
		"field2" : "oldvalue2"
	},
	"after" : {
		"field1" : "newvalue1",
		"field2" : "newvalue2"
	}
}
----

Many other Kafka Connect source connectors don't have the luxury of knowing this much about the changes, and instead use a simpler model where each message directly represents the after state of the row.
This is also what many sink connectors expect, and the Confluent JDBC Sink Connector is not different:

[source,json,indent=0]
----
{
	"field1" : "newvalue1",
	"field2" : "newvalue2"
}
----

While we think it's actually a great thing that Debezium CDC connectors provide as much detail as possible, we also make it easy for you to transform Debezium's _"envelope"_ format into the _"row"_ format that is expected by many other connectors.
Debezium provides a bridge between those two formats in a form of a https://cwiki.apache.org/confluence/display/KAFKA/KIP-66%3A+Single+Message+Transforms+for+Kafka+Connect[single message transform].
The `ExtractNewRecordState` transformation automatically extracts a new row record and thus effectively _flattens_ the complex record into a simple one consumable by other connectors.

You can use this SMT on the source connector to transform the message _before_ it is written to Kafka, or you can instead store the source connector's richer _"envelope"_ form of the message in Kafka and use this SMT on the sink connector to transform the message _after_ it is read from Kafka and before it is passed to the sink connector.
Both options work, and it just depends on whether you find the envelope form of the message useful for other purposes.

In our example we apply the SMT at the sink connector using these configuration properties:

[source,indent=0]
----
        "transforms": "unwrap",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
----

=== Delete records
When the Debezium connector detects a row is deleted, it creates two event messages: a _delete_ event and a _tombstone_ message.
The _delete_ message has an envelope with the state of the deleted row in the `before` field, and an `after` field that is `null`.
The _tombstone_ message contains same key as the _delete_ message, but the entire message value is `null`, and Kafka's log compaction utilizes this to know that it can remove any earlier messages with the same key.
A number of sink connectors, including the Confluent's JDBC Sink Connector, are not expecting these messages and will instead fail if they see either kind of message.
The `ExtractNewRecordState` SMT will by default filter out both _delete_ and _tombstone_ records, though you can change this if you're using the SMT and want to keep one or both of these kinds of messages.

=== Topic naming
Last but not least there is a difference in naming of topics.
Debezium uses fully qualified naming for target topics representing each table it manages.
The naming follows the pattern `<logical-name>.<database-name>.<table-name>`.
Kafka Connect JDBC Connector works with simple names `<table-name>`.

In more complex scenarios the user may deploy the https://kafka.apache.org/documentation/streams/[Kafka Streams] framework to establish elaborated routing between source and target routes.
In our example we will use a stock `RegexRouter` SMT that would route records created by Debezium into topics named according to JDBC Connector schema.
Again, we could use this SMT in either the source or sink connectors, but for this example we're going to use it in the source connector so we can choose the names of the Kafka topics where the records will be written.

[source,indent=0]
----
        "transforms": "route",
        "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
        "transforms.route.replacement": "$3"
----

== Example
Kick the tires and let's try our example!

First of all we need to deploy all components.
[source,bash,indent=0]
----
export DEBEZIUM_VERSION=0.6
docker-compose up
----

When all components are started we are going to register the JDBC Sink connector writing into PostgreSQL database:
[source,bash,indent=0]
----
curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @jdbc-sink.json
----

Using this registration request:

[source,json,indent=0]
----
{
    "name": "jdbc-sink",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "topics": "customers",
        "connection.url": "jdbc:postgresql://postgres:5432/inventory?user=postgresuser&password=<PASSWORD>",
        "transforms": "unwrap",                                                  (1)
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",(1)
        "auto.create": "true",                                                   (2)
        "insert.mode": "upsert",                                                 (3)
        "pk.fields": "id",                                                       (4)
        "pk.mode": "record_value"                                                (4)
    }
}
----

The request configures these options:

1. unwrapping Debezium's complex format into a simple one
2. automatically create target tables
3. insert a row if it does not exist or update an existing one
4. identify the primary key stored in Kafka's record value field

Then the source connector must be set up:

[source,bash,indent=0]
----
curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @source.json
----
{'fraction_non_alphanumeric': 0.09016184203808648, 'fraction_numerical': 0.009189768724153776, 'mean_word_length': 3.7998039696152905, 'pattern_counts': {'":': 40, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 20, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1557743', 'n_tokens_mistral': 5579, 'n_tokens_neox': 5296, 'n_words': 2344}