356 lines
17 KiB
Plaintext
356 lines
17 KiB
Plaintext
// Category: cdc-using
|
|
// Type: assembly
|
|
// ModuleID: configuring-debezium-connectors-to-use-avro-serialization
|
|
// Title: Configuring {prodname} connectors to use Avro serialization
|
|
[id="avro-serialization"]
|
|
= Avro Serialization
|
|
|
|
:toc:
|
|
:toc-placement: macro
|
|
:linkattrs:
|
|
:icons: font
|
|
:source-highlighter: highlight.js
|
|
|
|
toc::[]
|
|
|
|
ifdef::product[]
|
|
[IMPORTANT]
|
|
====
|
|
Using Avro to serialize record keys and values is a Technology Preview feature. Technology Preview features are not supported with Red Hat production service-level agreements (SLAs) and might not be functionally complete; therefore, Red Hat does not recommend implementing any Technology Preview features in production environments. This Technology Preview feature provides early access to upcoming product innovations, enabling you to test functionality and provide feedback during the development process. For more information about support scope, see link:https://access.redhat.com/support/offerings/techpreview/[Technology Preview Features Support Scope].
|
|
====
|
|
endif::product[]
|
|
|
|
A {prodname} connector works in the Kafka Connect framework to capture each row-level change in a database by generating a change event record. For each change event record, the {prodname} connector does the following:
|
|
|
|
. Applies configured transformations
|
|
. Serializes the record key and value into a binary form by using the configured link:https://kafka.apache.org/documentation/#connect_running[Kafka Connect converters]
|
|
. Writes the record to the correct Kafka topic
|
|
|
|
You can specify converters for each individual {prodname} connector instance. Kafka Connect provides a JSON converter that serializes the record keys and values into JSON documents. The default behavior is that the JSON converter includes the record's message schema, which makes each record very verbose. The {link-prefix}:{link-tutorial}[{name-tutorial}] shows what the records look like when both payload and schemas are included. If you want records to be serialized with JSON, consider setting the following connector configuration properties to `false`:
|
|
|
|
* `key.converter.schemas.enable`
|
|
* `value.converter.schemas.enable`
|
|
|
|
Setting these properties to `false` excludes the verbose schema information from each record.
|
|
|
|
Alternatively, you can serialize the record keys and values by using https://avro.apache.org/[Apache Avro]. The Avro binary format is compact and efficient. Avro schemas make it possible to ensure that each record has the correct structure. Avro's schema evolution mechanism enables schemas to evolve. This is essential for {prodname} connectors, which dynamically generate each record's schema to match the structure of the database table that was changed. Over time, change event records written to the same Kafka topic might have different versions of the same schema. Avro serialization makes it easier for change event record consumers to adapt to a changing record schema.
|
|
|
|
ifdef::community[]
|
|
To use Apache Avro serialization, you must deploy a schema registry that manages Avro message schemas and their versions.
|
|
Available options include the {registry-name-full} as well as the Confluent Schema Registry. Both are described here.
|
|
endif::community[]
|
|
|
|
ifdef::product[]
|
|
To use Apache Avro serialization, you must deploy a schema registry that manages Avro message schemas and their versions. For information about setting up this registry, see the documentation for {LinkServiceRegistryGetStart}[{registry-name-full}].
|
|
endif::product[]
|
|
|
|
// Type: concept
|
|
// Title: About the {registry}
|
|
[id="about-the-registry"]
|
|
== About the {registry-name-full}
|
|
|
|
ifdef::community[]
|
|
The link:https://github.com/Apicurio/apicurio-registry[{registry}] open-source project provides several components that work with Avro:
|
|
endif::community[]
|
|
|
|
ifdef::product[]
|
|
{LinkServiceRegistryGetStart}[{registry-name-full}] provides several components that work with Avro:
|
|
endif::product[]
|
|
|
|
* An Avro converter that you can specify in {prodname} connector configurations. This converter maps Kafka Connect schemas to Avro schemas. The converter then uses the Avro schemas to serialize the record keys and values into Avro's compact binary form.
|
|
|
|
* An API and schema registry that tracks:
|
|
+
|
|
** Avro schemas that are used in Kafka topics
|
|
** Where the Avro converter sends the generated Avro schemas
|
|
|
|
+
|
|
Since the Avro schemas are stored in this registry, each record needs to contain only a tiny _schema identifier_.
|
|
This makes each record even smaller. For an I/O bound system like Kafka, this means more total throughput for producers and consumers.
|
|
|
|
* Avro _Serdes_ (serializers and deserializers) for Kafka producers and consumers. Kafka consumer applications that you write to consume change event records can use Avro Serdes to deserialize the change event records.
|
|
|
|
To use the {registry} with {prodname}, add {registry} converters and their dependencies to the Kafka Connect container image that you are using for running a {prodname} connector.
|
|
|
|
[NOTE]
|
|
====
|
|
The {registry} project also provides a JSON converter. This converter combines the advantage of less verbose messages with human-readable JSON. Messages do not contain the schema information themselves, but only a schema ID.
|
|
====
|
|
|
|
// Type: concept
|
|
// Title: Overview of deploying a {prodname} connector that uses Avro serialization
|
|
[id="overview-of-deploying-a-debezium-connector-that-uses-avro-serialization"]
|
|
== Deployment overview
|
|
|
|
To deploy a {prodname} connector that uses Avro serialization, there are three main tasks:
|
|
|
|
ifdef::community[]
|
|
. Deploy an link:https://github.com/Apicurio/apicurio-registry[{registry-name-full}] instance.
|
|
endif::community[]
|
|
ifdef::product[]
|
|
. Deploy a link:{LinkServiceRegistryGetStart}[{registry-name-full} instance by following the instructions in {NameServiceRegistryGetStart}].
|
|
endif::product[]
|
|
|
|
ifdef::community[]
|
|
. Install the Avro converter from link:https://repo1.maven.org/maven2/io/apicurio/apicurio-registry-distro-connect-converter/{apicurio-version}/apicurio-registry-distro-connect-converter-{apicurio-version}-converter.tar.gz[the installation package] into Kafka Connect's _libs_ directory or directly into a plug-in directory.
|
|
endif::community[]
|
|
ifdef::product[]
|
|
. Install the Avro converter by downloading the {prodname} link:https://access.redhat.com/jbossnetwork/restricted/listSoftware.html?product=red.hat.integration&downloadType=distributions[Service Registry Kafka Connect] zip file and extracting it into the {prodname} connector's directory.
|
|
endif::product[]
|
|
|
|
. Configure a {prodname} connector instance to use Avro serialization by setting configuration properties as follows:
|
|
+
|
|
[source,options="nowrap"]
|
|
----
|
|
key.converter=io.apicurio.registry.utils.converter.AvroConverter
|
|
key.converter.apicurio.registry.url=http://apicurio:8080/api
|
|
key.converter.apicurio.registry.global-id=io.apicurio.registry.utils.serde.strategy.AutoRegisterIdStrategy
|
|
value.converter=io.apicurio.registry.utils.converter.AvroConverter
|
|
value.converter.apicurio.registry.url=http://apicurio:8080/api
|
|
value.converter.apicurio.registry.global-id=io.apicurio.registry.utils.serde.strategy.AutoRegisterIdStrategy
|
|
----
|
|
|
|
Internally, Kafka Connect always uses JSON key/value converters for storing configuration and offsets.
|
|
|
|
// Type: procedure
|
|
// Title: Deploying connectors that use Avro in {prodname} containers
|
|
// ModuleID: deploying-connectors-that-use-avro-in-debezium-containers
|
|
== Deploying with {prodname} containers
|
|
|
|
In your environment, you might want to use a provided {prodname} container to deploy {prodname} connectors that use Avro serializaion. Follow the procedure here to do that. In this procedure, you build a custom Kafka Connect container image for {prodname}, which uses the Avro converter.
|
|
|
|
.Prerequisites
|
|
|
|
* You have the required permissions on a Kafka cluster.
|
|
* You downloaded the {prodname} connector plug-in that you want to deploy with Avro serialization.
|
|
|
|
.Procedure
|
|
|
|
. Deploy an instance of {registry}.
|
|
+
|
|
ifdef::community[]
|
|
The following example uses a non-production, in-memory, {registry} instance:
|
|
+
|
|
[source,subs="attributes+"]
|
|
----
|
|
docker run -it --rm --name apicurio \
|
|
-p 8080:8080 apicurio/apicurio-registry-mem:{apicurio-version}
|
|
----
|
|
endif::community[]
|
|
ifdef::product[]
|
|
Follow the instructions in
|
|
{LinkServiceRegistryGetStart}#installing-registry-operatorhub[NameServiceRegistryGetStart, Installing Service Registry from the OpenShift OperatorHub].
|
|
endif::product[]
|
|
|
|
. Build a {prodname} container image that contains the Avro converter:
|
|
+
|
|
ifdef::community[]
|
|
.. Copy link:https://github.com/debezium/debezium-examples/blob/master/tutorial/debezium-with-apicurio/Dockerfile[`Dockerfile`] to a convenient location. This file has the following content:
|
|
+
|
|
[listing,subs="attributes",options="nowrap"]
|
|
----
|
|
ARG DEBEZIUM_VERSION
|
|
FROM debezium/connect:$DEBEZIUM_VERSION
|
|
ENV KAFKA_CONNECT_DEBEZIUM_DIR=$KAFKA_CONNECT_PLUGINS_DIR/debezium-connector-mysql
|
|
ENV APICURIO_VERSION={apicurio-version}
|
|
|
|
RUN cd $KAFKA_CONNECT_DEBEZIUM_DIR &&\
|
|
curl https://repo1.maven.org/maven2/io/apicurio/apicurio-registry-distro-connect-converter/$APICURIO_VERSION/apicurio-registry-distro-connect-converter-$APICURIO_VERSION-converter.tar.gz | tar xzv
|
|
----
|
|
|
|
.. Run the following command:
|
|
+
|
|
[source,subs="attributes+"]
|
|
----
|
|
docker build --build-arg DEBEZIUM_VERSION={debezium-docker-label} -t debezium/connect-apicurio:{debezium-docker-label} .
|
|
----
|
|
endif::community[]
|
|
ifdef::product[]
|
|
.. Download the link:https://access.redhat.com/jbossnetwork/restricted/listSoftware.html?product=red.hat.integration&downloadType=distributions[{registry} Kafka Connect] zip file.
|
|
.. Extract the content into the directory that contains the {prodname} connector that you are configuring to use Avro serialization.
|
|
.. Create a custom image for Kafka Connect. See link:{LinkCDCInstallOpenShift}[{NameCDCInstallOpenShift}, Creating a container image from the Kafka Connect base image] for an example of how to do this. Start with the `Dockerfile` in that example. Then add the {registry} converters to the connector directories.
|
|
endif::product[]
|
|
|
|
. Run the newly built Kafka Connect image, configuring it so it uses the Avro converter:
|
|
+
|
|
ifdef::community[]
|
|
[source,subs="attributes+"]
|
|
----
|
|
docker run -it --rm --name connect \
|
|
--link zookeeper:zookeeper \
|
|
--link kafka:kafka \
|
|
--link mysql:mysql \
|
|
--link apicurio:apicurio \
|
|
-e GROUP_ID=1 \
|
|
-e CONFIG_STORAGE_TOPIC=my_connect_configs \
|
|
-e OFFSET_STORAGE_TOPIC=my_connect_offsets \
|
|
-e KEY_CONVERTER=io.apicurio.registry.utils.converter.AvroConverter \
|
|
-e VALUE_CONVERTER=io.apicurio.registry.utils.converter.AvroConverter \
|
|
-e CONNECT_KEY_CONVERTER=io.apicurio.registry.utils.converter.AvroConverter \
|
|
-e CONNECT_KEY_CONVERTER_APICURIO.REGISTRY_URL=http://apicurio:8080 \
|
|
-e CONNECT_KEY_CONVERTER_APICURIO.REGISTRY_GLOBAL-ID=io.apicurio.registry.utils.serde.strategy.AutoRegisterIdStrategy \
|
|
-e CONNECT_VALUE_CONVERTER=io.apicurio.registry.utils.converter.AvroConverter \
|
|
-e CONNECT_VALUE_CONVERTER_APICURIO_REGISTRY_URL=http://apicurio:8080 \
|
|
-e CONNECT_VALUE_CONVERTER_APICURIO_REGISTRY_GLOBAL-ID=io.apicurio.registry.utils.serde.strategy.AutoRegisterIdStrategy \
|
|
-p 8083:8083 debezium/connect-apicurio:{debezium-docker-label}
|
|
----
|
|
endif::community[]
|
|
ifdef::product[]
|
|
|
|
.. Follow the steps in the link:{LinkCDCGettingStarted}#deploying-kafka-connect[example of deploying Kafka Connect] in {NameCDCGettingStarted}.
|
|
.. Open the `examples/kafka-connect/kafka-connect-s2i-single-node-kafka.yaml` file that you used to deploy Kafka Connect.
|
|
+
|
|
Before you can create the connector instance,
|
|
you must first enable connector resources in the `KafkaConnectS2I` Custom Resource (CR).
|
|
|
|
.. In the `metadata.annotations` section, enable Kafka Connect to use connector resources.
|
|
+
|
|
.kafka-connect-s2i-single-node-kafka.yaml
|
|
[source,yaml,options="nowrap"]
|
|
----
|
|
apiVersion: kafka.strimzi.io/v1beta1
|
|
kind: KafkaConnectS2I
|
|
metadata:
|
|
name: my-connect-cluster
|
|
annotations:
|
|
strimzi.io/use-connector-resources: "true"
|
|
spec:
|
|
...
|
|
----
|
|
|
|
.. Apply the updated `kafka-connect-s2i-single-node-kafka.yaml` file to update the `KafkaConnectS2I` CR:
|
|
+
|
|
`oc apply -f kafka-connect-s2i-single-node-kafka.yaml`
|
|
|
|
.. In the Kafka Connect CR that defines the connector, add the properties that are required by the Avro converter. The CR looks like this:
|
|
+
|
|
.inventory-connector.yaml
|
|
[source,yaml,options="nowrap"]
|
|
----
|
|
apiVersion: kafka.strimzi.io/v1beta1
|
|
kind: KafkaConnector
|
|
metadata:
|
|
name: inventory-connector
|
|
labels:
|
|
strimzi.io/cluster: my-connect-cluster
|
|
spec:
|
|
class: io.debezium.connector.mysql.MySqlConnector
|
|
tasksMax: 1
|
|
config:
|
|
database.hostname: mysql
|
|
database.port: 3306
|
|
database.user: debezium
|
|
database.password: dbz
|
|
database.server.id: 184054
|
|
database.server.name: dbserver1
|
|
database.whitelist: inventory
|
|
database.history.kafka.bootstrap.servers: my-cluster-kafka-bootstrap:9092
|
|
database.history.kafka.topic: schema-changes.inventory
|
|
key.converter: io.apicurio.registry.utils.converter.AvroConverter
|
|
key.converter.apicurio.registry.url: http://apicurio:8080/api
|
|
key.converter.apicurio.registry.global-id: io.apicurio.registry.utils.serde.strategy.AutoRegisterIdStrategy
|
|
value.converter: io.apicurio.registry.utils.converter.AvroConverter
|
|
value.converter.apicurio.registry.url: http://apicurio:8080/api
|
|
value.converter.apicurio.registry.global-id: io.apicurio.registry.utils.serde.strategy.AutoRegisterIdStrategy
|
|
----
|
|
|
|
.. Apply the connector instance:
|
|
+
|
|
`$ oc apply -f inventory-connector.yaml`
|
|
+
|
|
This registers `inventory-connector` and the connector starts to run against the `inventory` database.
|
|
|
|
.. To verify that the connector was created and has started to monitor the database, follow the steps at the end of the example procedure in link:{LinkCDCGettingStarted}#creating-connector-monitor-inventory-database[NameCDCGettingStarted].
|
|
endif::product[]
|
|
|
|
// Type: concept
|
|
// Title: About Avro name requirements
|
|
// ModuleID: about-avro-name-requirements
|
|
[[avro-naming]]
|
|
== Naming
|
|
|
|
As stated in the Avro link:https://avro.apache.org/docs/current/spec.html#names[documentation], names must adhere to the following rules:
|
|
|
|
* Start with `[A-Za-z_]`
|
|
* Subsequently contains only `[A-Za-z0-9_]` characters
|
|
|
|
{prodname} uses the column's name as the basis for the corresponding Avro field.
|
|
This can lead to problems during serialization if the column name does not also adhere to the Avro naming rules.
|
|
Each {prodname} connector provides a configuration property, `sanitize.field.names` that you can set to `true` if you have columns that do not adhere to Avro rules for names. Setting `sanitize.field.names` to `true` allows serialization of non-conformant fields without having to actually modify your schema.
|
|
|
|
ifdef::community[]
|
|
[id="confluent-schema-registry"]
|
|
== Confluent Schema Registry
|
|
|
|
There is an alternative https://github.com/confluentinc/schema-registry[schema registry] implementation provided by Confluent. The configuration is slightly different.
|
|
|
|
. In your {prodname} connector configuration, specify the following properties:
|
|
+
|
|
[source]
|
|
----
|
|
key.converter=io.confluent.connect.avro.AvroConverter
|
|
key.converter.schema.registry.url=http://localhost:8081
|
|
value.converter=io.confluent.connect.avro.AvroConverter
|
|
value.converter.schema.registry.url=http://localhost:8081
|
|
----
|
|
|
|
. Deploy an instance of the Confluent Schema Registry:
|
|
+
|
|
[source]
|
|
----
|
|
docker run -it --rm --name schema-registry \
|
|
--link zookeeper \
|
|
-e SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL=zookeeper:2181 \
|
|
-e SCHEMA_REGISTRY_HOST_NAME=schema-registry \
|
|
-e SCHEMA_REGISTRY_LISTENERS=http://schema-registry:8081 \
|
|
-p 8181:8181 confluentinc/cp-schema-registry
|
|
----
|
|
|
|
. Run a Kafka Connect image configured to use Avro:
|
|
+
|
|
[source,subs="attributes"]
|
|
----
|
|
docker run -it --rm --name connect \
|
|
--link zookeeper:zookeeper \
|
|
--link kafka:kafka \
|
|
--link mysql:mysql \
|
|
--link schema-registry:schema-registry \
|
|
-e GROUP_ID=1 \
|
|
-e CONFIG_STORAGE_TOPIC=my_connect_configs \
|
|
-e OFFSET_STORAGE_TOPIC=my_connect_offsets \
|
|
-e KEY_CONVERTER=io.confluent.connect.avro.AvroConverter \
|
|
-e VALUE_CONVERTER=io.confluent.connect.avro.AvroConverter \
|
|
-e CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL=http://schema-registry:8081 \
|
|
-e CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL=http://schema-registry:8081 \
|
|
-p 8083:8083 debezium/connect:{debezium-docker-label}
|
|
----
|
|
|
|
. Run a console consumer that reads new Avro messages from the `db.myschema.mytable` topic and decodes to JSON:
|
|
+
|
|
[source,subs="attributes"]
|
|
----
|
|
docker run -it --rm --name avro-consumer \
|
|
--link zookeeper:zookeeper \
|
|
--link kafka:kafka \
|
|
--link mysql:mysql \
|
|
--link schema-registry:schema-registry \
|
|
debezium/connect:{debezium-docker-label} \
|
|
/kafka/bin/kafka-console-consumer.sh \
|
|
--bootstrap-server kafka:9092 \
|
|
--property print.key=true \
|
|
--formatter io.confluent.kafka.formatter.AvroMessageFormatter \
|
|
--property schema.registry.url=http://schema-registry:8081 \
|
|
--topic db.myschema.mytable
|
|
----
|
|
|
|
== Getting More Information
|
|
|
|
link:/blog/2016/09/19/Serializing-Debezium-events-with-Avro/[This post] from the {prodname} blog
|
|
describes the concepts of serializers, converters, and other components, and discusses the advantages of using Avro.
|
|
Some Kafka Connect converter details have slightly changed since that post was written.
|
|
|
|
For a complete example of using Avro as the message format for {prodname} change data events,
|
|
see https://github.com/debezium/debezium-examples/tree/master/tutorial#using-mysql-and-the-avro-message-format[MySQL and the Avro message format].
|
|
|
|
endif::community[]
|