Darko Kojović

In the previous blog we explained what Kafka Connect is, what we can do with it, and what are some of the most important components that we can use to establish a reliable and fault-tolerant big data pipeline. In this blog, we will deploy a small-scale but production-ready Kafka Connect cluster where we will take Avro and Protobuf messages from different topics and store them in Snowflake tables. We will use Schema Registry to validate our Avro and Protobuf schemas. In theory, it seems like a straightforward process but in practice, it can get a bit complex especially if you are setting up Kafka Connect for the first time. So many properties can be easily overlooked, but precisely those properties can save you from losing your precious Kafka messages. We will go over the whole process and show you how we can set up a production-ready pipeline from scratch. Environment setup Kubernetes cluster* Kafka cluster Schema registry Snowflake account *Kubernetes cluster is not required, but if you plan to set up Kafka Connect somewhere else (like EC2, VMs), you must make some adjustments. All the above services (besides Snowflake) can be deployed locally or you can use managed services. To make this blog less complex since the main focus is on Kafka Connect, (for our example), we will use Confluent Platform for the Kafka cluster and schema registry. We understand that not everybody can use Confluent Platform for commercial projects, and you do not have to. The process is not tied to Confluent, which can be achieved without using any Confluent products; instead, it can be done using open-source software. For reference, our company is not using anything related to Confluent, not even the schema registry or converters. If you want to use Confluent Platform to follow along, there is a 30 day free trial that is more than enough to replicate everything from this blog. The same 30 day trial exists in Snowflake too. To confirm that it also works without the Confluent Platform we also deployed Kafka using Koperator on our Kubernetes cluster, and the setup was the same. For the schema registry, you can try out Karapace. Setting up Kafka Connect to store messages in Snowflake To make our example more straightforward, we will imagine that we are a sushi restaurant that uses Kafka, Kubernetes and Snowflake to manage their customers, orders and reviews (this is probably not the best approach but is easier to understand).Even though we will use a small sample dataset, the same setup will work with any dataset size. Our company is using a similar pipeline in production to process millions of records each day, and it works without any issues. Kafka Connect clusters are horizontally scalable, which means that you will only need to increase the replica count to support larger workloads. Our application will have three topics: two will use Protobuf, and one will use Avro. We wanted to cover multiple data serialization formats to demonstrate how to handle various different Kafka Connect clusters. The diagram that we will follow to set up our Kafka to Snowflake pipeline is below: We will be using Kubernetes, but if you want to deploy it somewhere else, the Kafka Connect setup is the same; the only difference is how you want to deploy it. Feel free to use this Docker image dksadx/kafka-connect-snowflake, or if you want to deploy it on bare metal, you can use this Dockerfile as a guide. Setting up the demo environment (optional) This section is optional; if you already have Kafka, Kubernetes, and Snowflake set up, feel free to skip this and go straight to the Configuring Kafka Connect section. If you want to follow along, we need to set up a demo environment first. Creating a Kubernetes cluster We will use Podman with Kind to set up a local Kubernetes cluster. You can use any other tool to set up your cluster or deploy a managed cluster via your preferred cloud provider. # Install Podman, QEMO, Kind and kubectl (MacOS only) $ brew install podman qemu kind kubectl helm # Create and start a VM for the cluster (increase resources if deploying Kafka) $ podman machine init --cpus 4 --disk-size 30 --memory 8192 --now demo-vm # Create a Kubernetes cluster with Kind $ KIND_EXPERIMENTAL_PROVIDER=podman kind create cluster --name kafka_connect_demo # Set the context $ kubectl config set-context kafka_connect_demo Check if everything works correctly: $ kubectl get all # NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE # service/kubernetes ClusterIP 10.96.0.1 <none> 443/TCP 2m48s Setting up Kafka on Kubernetes If you do not have a Kafka cluster, you can create one for free via the Confluent Platform. After creating the cluster, create three topics called avro-customers, proto-orders, and proto-reviews. If you do not want to use the Confluent Platform and instead want to deploy it locally on the Kubernetes cluster we previously created, the easiest way to set it up is via the Koperator Helm chart. To set up the latest and most up-to-date version of the chart, use their guide, which can be found in the README.md here. After the setup, we can create three topics for our sushi demo app. kubectl create -n kafka -f - <<EOF apiVersion: kafka.banzaicloud.io/v1alpha1 kind: KafkaTopic metadata: name: avro-customers spec: clusterRef: name: kafka name: customers partitions: 1 replicationFactor: 1 config: "retention.ms": "604800000" "cleanup.policy": "delete" EOF Do the same for the other two topics: proto-reviews, and proto-orders. Setting up the Snowflake database For storing Kafka messages, we will use Snowflake, and thus, we will need a Snowflake account. If you do not have an account, you can sign up for a free 30-day trial on their page: www.snowflake.com. After creating an account, we will need to create a warehouse. If you are not familiar with Snowflake, warehouses are basically compute instances in Snowflake. You can create a warehouse by going to the Admin panel and clicking the blue + WAREHOUSE button in the top right corner. Choose XS size for this demo. After creating a warehouse, go to Projects -> Worksheets -> SQL Worksheet. Now, we need to configure Snowflake so that we can actually use the Kafka Connect connector. There are few steps that we have to do: Create a database and schema: CREATE DATABASE kafka_connect_demo; CREATE SCHEMA kafka_connect_demo.sushi_restaurant;   Create user, role and configure all the privileges: -- Use a role that can create and manage roles and privileges. USE ROLE accountadmin; -- Create a Snowflake role with the privileges to work with the connector. CREATE ROLE kafka_connector_role_sushi; -- Create a Snowflake user that will be used by the kafka connect connector to access Snowflake. CREATE USER IF NOT EXISTS kafka_connect_sushi_user PASSWORD = "my_very_strong_password" LOGIN_NAME = "kafka_connect_sushi_user" DEFAULT_ROLE = "kafka_connector_role_sushi" DEFAULT_SECONDARY_ROLES = ('ALL'); -- Grant privileges on the database. GRANT USAGE ON DATABASE kafka_connect_demo TO ROLE kafka_connector_role_sushi; -- Grant privileges on the schema. GRANT USAGE ON SCHEMA kafka_connect_demo.sushi_restaurant TO ROLE kafka_connector_role_sushi; GRANT CREATE TABLE ON SCHEMA kafka_connect_demo.sushi_restaurant TO ROLE kafka_connector_role_sushi; GRANT CREATE STAGE ON SCHEMA kafka_connect_demo.sushi_restaurant TO ROLE kafka_connector_role_sushi; GRANT CREATE PIPE ON SCHEMA kafka_connect_demo.sushi_restaurant TO ROLE kafka_connector_role_sushi; -- Grant the custom role to an existing user. GRANT ROLE kafka_connector_role_sushi TO USER kafka_connect_sushi_user; -- Grant the custom role to your user (assuming you are using ACCOUNTADMIN) GRANT ROLE kafka_connector_role_sushi TO USER ACCOUNTADMIN; -- Set the custom role as the default role for the user. -- If you encounter an 'Insufficient privileges' error, verify the role that has the OWNERSHIP privilege on the user. ALTER USER kafka_connect_sushi_user SET DEFAULT_ROLE = kafka_connector_role_sushi; -- Grant privileges on a warehouse to our role. GRANT ALL PRIVILEGES ON WAREHOUSE kafkaconnect TO ROLE kafka_connector_role_sushi;   Generate an RSA key pair. If you are unfamiliar with how to do that, you can use this guide: Using key pair authentication & key rotation. Add the public key to the user you created in step 2, in our case the user is called kafka_connect_sushi_user. You must remove all the new lines and also the headers and footers (-----BEGIN PUBLIC KEY----- , -----END PUBLIC KEY-----) from the public key. -- Add RSA public key to user. This is how your public key needs to look like, without headers and new lines. ALTER USER kafka_connect_sushi_user SET RSA_PUBLIC_KEY='MIIBIj...';   While still signed in to Snowflake, go to the home pages and save the url. The url should look like this: https://app.snowflake.com/<orgname>/<accountname>/worksheets. We will need this information when setting up Kafka Connect.   Setting up Schema Registry Having a schema registry is not required, and you can do the whole setup without it but I highly recommend using it. If you are working or planning to work on a larger project, having a schema registry can save you a lot of trouble. If you do not plan to use schema registry, skip this section and remove all config properties that have schema in its name and you should be good to go. To set up a schema registry there are two choices: sign up to Confluent Platform and use schema registry as a service, host your own. Whichever option you choose, the rest will be the same; only the self-hosted one will require additional steps. To reduce the complexity of the blog, we will use Confluent Platform for this demo. If you choose to use Confluent Platform, go to their site, create an account, and start a 30-day free trial. After finishing all the required steps, your environment and cluster should be ready. The only thing that we need to do here is find the endpoint url and generate an API key. You can do that by clicking on your environment and in the right sidebar you can find the endpoint. Below the endpoint, you can generate an API key. Save the above info because we will need it in the next step. Configuring Kafka Connect After everything is set up and deployed, we are ready for the main part, configuring and deploying Kafka Connect clusters. For our example, we will need three clusters since we have avro, protobuf, and string converters. We will configure and deploy Kafka Connect in a distributed mode because it is the most reliable and fault-tolerant. In most cases, you will want to use this mode. There are two configuration files that we need to edit: connect-distributed.properties - a file that configures the Kafka Connect cluster, snowflake.json - file that configures the Snowflake connector. I will not go into too much detail about what all properties mean because you can find that in the official documentation. I will go over the most interesting ones for our use case. connect-distributed.properties This file configures the Kafka Connect cluster, it does not configure any connectors. We will add connectors after the cluster is running and in a healthy state. Since we are deploying our cluster in distributed mode, we will need three additional topics that will be used to store the configs, statuses, and offsets for the clusters. This is what the configuration for our avro cluster looks like: # Unique name for the cluster, used in forming the Connect cluster group. Note that this must not conflict with consumer group IDs group.id=kafka-connect-avro # Topic to use for storing offsets. This topic should have many partitions and be replicated and compacted. # Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create # the topic before starting Kafka Connect if a specific topic configuration is needed. offset.storage.topic=kafka-connect-avro-offsets offset.storage.replication.factor=3 offset.storage.partitions=3 # Topic to use for storing connector and task configurations; note that this should be a single partition, highly replicated, # and compacted topic. Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create # the topic before starting Kafka Connect if a specific topic configuration is needed. config.storage.topic=kafka-connect-avro-configs config.storage.replication.factor=3 config.storage.partitions=1 # Topic to use for storing statuses. This topic can have multiple partitions and should be replicated and compacted. # Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create # the topic before starting Kafka Connect if a specific topic configuration is needed. status.storage.topic=kafka-connect-avro-statuses status.storage.replication.factor=3 status.storage.partitions=3 You can create the three above topics yourself or let Kafka Connect create it for you. To configure the schema registry, we will use the information that we got from the previous step: value.converter.basic.auth.credentials.source=USER_INFO value.converter.basic.auth.user.info=YOUR_API_KEY:YOUR_API_SECRET value.converter.expected.schema.name=customers value.converter.schema.registry.url=https://YOUR_ACCOUNT.confluent.cloud The most interesting properties for us are the converter properties because if those are wrong, we will not be able to serialize or deserialize messages correctly. We need to define which converters we want to use for both keys and values. Usually you can use StringConverter for keys but for values you need specific converters. There are many converters out there, some are officially supported and some are custom made for specific use cases by the community. An example of such a converter is registryless-avro-converter, which is an avro converter that does not depend on the Confluent Schema Registry. We will use 3 types of converters in our example: Avro - io.confluent.connect.avro.AvroConverter Protobuf - io.confluent.connect.protobuf.ProtobufConverter String - org.apache.kafka.connect.storage.StringConverter key.converter=org.apache.kafka.connect.storage.StringConverter # Using avro converter value.converter=io.confluent.connect.avro.AvroConverter snowflake.json This file configures the Snowflake connector for Kafka Connect. It is a JSON file with all the secrets and configurations required for the connector to function. Since this is a connector configuration file, we do not specify it while starting the cluster; instead, we need to send a POST request with the file content as payload to the Kafka Connect endpoint once the cluster is up and running. You can send the request from anywhere; it does not need to be from the container or pod, but I recommend that you do not expose the Kafka Connect endpoint externally if you really do not need to, mainly for security reasons. Having the snowflake.json file along with the connect-distributed.properties file can be beneficial, as it allows us to recreate the connector if something unexpected happens automatically. If you are running the cluster for months, something may break the cluster, and having a backup solution will be a lifesaver in those moments. It rarely happens, but occurs sometimes, so configuring liveness and startup probes to take care of this is a big plus. You can find detailed instructions on configuring a Snowflake connector in their official documentation here. Most of the configuration is self-explanatory but I will point out a few fields that may cause problems: name Name of the connector. This name will be part of the url, and it will be used when performing any actions on the connector, e.g., updating the config properties. If you plan to have multiple connectors on a single cluster, make sure they have different names. snowflake.url.name Format for the url is <orgname>-<accountname>.snowflakecomputing.com:443. You can find the <orgname> and <accountname> in the url when you log in to your account via the browser. The url should be similar to this: https://app.snowflake.com/kalsvgw/ab20210/worksheets where <orgname> is kalsvgw and <accountname> is ab20210. snowflake.private.key This is the encrypted RSA key that we generated in the section where we did the Snowflake setup. Make sure that you remove all line breaks, whitespace and headers/footers before pasting the key. snowflake.private.key.passphrase This is the passphrase used while creating the key. Deploying Kafka Connect clusters We are almost ready to deploy our Kafka Connect clusters once the configurations for Kafka Connect and Snowflake connector are ready. Starting the Kafka Connect in distributed mode is as simple as running the connect-distributed.sh script that comes with every Kafka installation and providing the configuration properties as the first argument: $ ./connect-distributed.sh connect-distributed.properties Keep in mind that you need different dependencies (jar files) for different connectors and converters. To see which dependencies are required for our example you can check this Dockerfile. If everything is configured correctly, we will see that three new topics for offsets, statuses, and configs appeared in our Kafka cluster. These topics are used by the cluster instances/replicas to communicate with each other and distribute the load. After the cluster is up and there are no errors in the logs, we can add connectors. To add a new connector, we need to send a POST request to the cluster's endpoint: curl -X POST -H "Content-Type: application/json" --data @snowflake.json http://localhost:8083/connectors We should see messages arriving in our Snowflake tables if everything is configured correctly. Since we will be deploying the cluster in Kubernetes for our sushi restaurant example, we prepared a Helm chart that will take care of most of the setup. We only have to configure some of the cluster and connector properties like bootstrap servers and credentials. First we need to get the chart   git clone https://github.com/DKSadx/kafka-connect-blog.git cd deploy/helm You will see three types of configuration files in the configs directory, connect-distributed-<TYPE>.properties, snowflake-<TYPE>.json, and config.yaml. Fill out the empty fields with your Kafka and Snowflake information in all of those configs. If you are confused about what config.yaml is, that is a configuration file that is used by our simple demo Go app called sushi-restaurant-kafka-demo-app. The purpose of the app is to read some example dummy messages from app/data/ directory and publish those messages to corresponding topics specified in config.yaml. Besides topics, the configuration file also contains information about the Kafka cluster so that it can connect and publish those messages. After everything is configured, we are ready to install the helm chart: helm install sushi-restaurant . Let's check the pods and wait for everything to be in RUNNING state: $ kubectl get pods NAME READY STATUS RESTARTS AGE demo-app-dqw6n 0/1 Completed 0 9s sushi-restaurant-customers-avro-0-c8d7c8689-brmm7 1/1 Running 0 9s sushi-restaurant-customers-avro-0-c8d7c8689-sh4h9 1/1 Running 0 9s sushi-restaurant-dlq-string-2-56f7cf687-58qb9 1/1 Running 0 9s sushi-restaurant-dlq-string-2-56f7cf687-kwhpg 1/1 Running 0 9s sushi-restaurant-reviews-orders-protobuf-1-6546fd87fb-5882s 0/1 Running 0 9s sushi-restaurant-reviews-orders-protobuf-1-6546fd87fb-9dskd 0/1 Running 0 9s When you inspect the logs of the demo app, they need to look like this. If that is the case, that means that the messages are being published to our topics: $ kubectl get logs demo-app-dqw6n 2024/09/02 12:08:40 Created Producer rdkafka#producer-1 2024/09/02 12:08:40 Publishing orders to proto-orders topic 2024/09/02 12:08:40 Publishing customers to avro-customers topic 2024/09/02 12:08:40 Publishing reviews to proto-reviews topic 2024/09/02 12:08:41 Delivered message to topic proto-reviews [1] at offset 0 2024/09/02 12:08:41 Delivered message to topic proto-orders [1] at offset 0 2024/09/02 12:08:41 Delivered message to topic avro-customers [0] at offset 0 2024/09/02 12:08:41 Delivered message to topic proto-orders [1] at offset 1 2024/09/02 12:08:41 Delivered message to topic avro-customers [1] at offset 0 2024/09/02 12:08:41 Delivered message to topic proto-orders [1] at offset 2 When you now check logs of any of the Kafka Connect pods, if the log looks similar to this, then it means that messages are successfully being consumed from topics and pushed to Snowflake: $ kubectl logs sushi-restaurant-customers-avro-0-c8d7c8689-6gqkj 2024-09-02 12:11:32,784] INFO [snowflake_sushi_customers|task-2] [SF_INGEST] Channel=CUSTOMERS_2 created for table=CUSTOMERS (net.snowflake.ingest.streaming.internal.SnowflakeStreamingIngestChannelInternal:58) [2024-09-02 12:11:33,140] INFO [snowflake_sushi_customers|task-0] [SF_KAFKA_CONNECTOR] Fetched offsetToken for channelName:KAFKA_CONNECT_DEMO.SUSHI_RESTAURANT.CUSTOMERS.CUSTOMERS_0, offset:1 (com.snowflake.kafka.connector.internal.streaming.TopicPartitionChannel:46) [2024-09-02 12:11:33,168] INFO [snowflake_sushi_customers|task-0] [SF_KAFKA_CONNECTOR] task opened with 1 partitions, execution time: 2955 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask:46) [2024-09-02 12:11:33,169] INFO [snowflake_sushi_customers|task-0] [Consumer clientId=connector-consumer-snowflake_sushi_customers-0, groupId=avro-customers] Seeking to offset 2 for partition customers-0 (org.apache.kafka.clients.consumer.KafkaConsumer:1585) [2024-09-02 12:11:33,469] INFO [snowflake_sushi_customers|task-2] [SF_KAFKA_CONNECTOR] Fetched offsetToken for channelName:KAFKA_CONNECT_DEMO.SUSHI_RESTAURANT.CUSTOMERS.CUSTOMERS_2, offset:2 (com.snowflake.kafka.connector.internal.streaming.TopicPartitionChannel:46) [2024-09-02 12:11:33,471] INFO [snowflake_sushi_customers|task-2] [SF_KAFKA_CONNECTOR] task opened with 1 partitions, execution time: 3255 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask:46) [2024-09-02 12:11:33,471] INFO [snowflake_sushi_customers|task-2] [Consumer clientId=connector-consumer-snowflake_sushi_customers-2, groupId=avro-customers] Seeking to offset 3 for partition customers-2 (org.apache.kafka.clients.consumer.KafkaConsumer:1585) [2024-09-02 12:11:36,038] INFO [snowflake_sushi_customers|task-0] [SF_KAFKA_CONNECTOR] Fetched offsetToken for channelName:KAFKA_CONNECT_DEMO.SUSHI_RESTAURANT.CUSTOMERS.CUSTOMERS_0, offset:1 (com.snowflake.kafka.connector.internal.streaming.TopicPartitionChannel:46) You are probably wondering, how does it work if we did not add the Snowflake connector. The reason for this is that we automated the process using Startup Probes and this script. The script will check if the connector exists, and if not, it will add it automatically. Now, if we go to Snowflake and check any of the new tables, you will see the example messages there:   Dead letter queue to the rescue Great, the pipeline works, but what happens if somebody sends a message that was serialized using a different converter, e.g., sends a protobuf message on an avro topic. We can not ignore those messages, maybe those are some really important transactions that were sent by mistake. To handle such cases, we need a dead letter queue or DLQ. If you remember from the previous blog or check the example configuration file snowflake.json in the Helm chart, you will see that DLQ properties are already configured: "errors.deadletterqueue.topic.name": "kafka_connect_dlq", "errors.deadletterqueue.topic.replication.factor": 3, "errors.deadletterqueue.context.headers.enable": true You will also see one more property that is not directly related to DLQ but is important to set it up correctly. "errors.tolerance": "all" By default, error.tolerance is set to none, which means that if a message can not be deserialized, the Kafka Connect instance will throw an error and stop. For some use cases, that can be exactly what we want, but for some, we can not afford to stop the whole pipeline if someone publishes a message to a wrong topic. In such cases it is good to have a Kafka Connect cluster specifically for DLQ with the property error.tolerance set to all. With that approach you have a dedicated cluster that will take all the failed messages that went to the DLQ and push those messages to Snowflake as they are. Now, we can continue consuming messages while also saving the failed messages to be analyzed later. For production environments, you will want to configure alerts on the Snowflake side so that you will be notified if new messages are added to the DLQ table. Conclusion When faced with the challenge of transferring messages from Kafka to another destination, in our case, it was Snowflake, Kafka Connect will be the best approach for most cases.  We used Snowflake as an example since we are using it heavily on our projects and it is becoming a go-to platform when it comes to big data pipelines. Snowflake is just one sink connector, but there are hundreds more. If you want to take messages from Kafka topics and upload them somewhere or vice-versa, there is a good chance that someone already wrote a connector for that. Having written custom consumers and discovering bugs when least expecting them is never a good feeling. That is one of the reasons why using a battle-tested tool with many users and a large community behind it is, in most cases, the way to go.

Every year, zettabytes of data are transferred over the Internet. Managing, processing, and storing the data can be a really complex task and requires cutting-edge tools. Many tools can be used, but two technologies that are almost essential to every modern data pipeline: Kafka and Snowflake. If you are planning to stream a large amount of data, Kafka is usually the best choice. Scalability, reliability, low latency and popularity are reasons why Kafka stands out. Kafka is by far the most popular event streaming platform and a go-to for many cases when it comes to streaming data. Since you need to process and store the data somewhere, Snowflake is one of the top contenders for that.  If you are wondering why Snowflake is so popular, look at the user satisfaction - it’s over the roof. Snowflake took most of the drawbacks and oversights that other data storage solutions had and addressed it. Probably the most important feature is having the storage and compute separately scaled.  What is Kafka Connect, and why did we decide to use it? Kafka Connect is a tool for scalably and reliably streaming data between Apache Kafka and other data systems. It provides streaming integration between Kafka and other data sources or sinks, enabling seamless data movement and transformation. Kafka Connect uses connectors to ingest data into Kafka topics or export data from Kafka topics to external systems. Kafka Connect is made for use cases like this. Installing Kafka Connect and choosing the correct mode If you search for Kafka Connect, you will see that there is no download link. That is because Kafka Connect is not a standalone tool; it is bundled with the main Kafka installation. If you communicated with Kafka from the command line before, you probably used one of the scripts from the bin directory like kafka-topics.sh. Two more scripts that are used for Kafka Connect are in the same directory as the kafka-topics.sh script. Those two scripts are for two different modes, standalone and distributed: connect-standalone.sh - Only runs one node of Kafka Connect and saves all the configs, offsets, and statuses locally. If the node goes down, all Kafka Connect processes are down. This is okay for testing but should not be used in production as it is neither reliable nor fault-tolerant.  connect-distributed.sh - Runs a cluster of Kafka Connect nodes. Saves all configs, offsets and statuses in specific Kafka topics designated for Kafka Connect. If one node goes down, another node will take over the workload from that node. The connect-standalone.sh script can be used for testing, but connect-distributed.sh should be used for production. Kafka Connect components There are a few Kafka Connect components, but the most important for us are: Connectors (sink and source) Converters (including transforms) Dead Letter Queue (DLQ) If you are new to Kafka Connect, you may be confused about the difference between connectors and converters.  Connectors Connectors, as the name says, connect your Kafka Connect cluster to a specific source/destination. In our case, that will be Snowflake, but it can also be HDFS, S3, Elasticsearch, and many more. There are two types of connectors: Source Connectors - Take the data from a source and send it to Kafka topic(s). In the diagram below, we can see how it would look if we wanted to take files from S3 and publish them to a Kafka topic. Sink Connectors - Take the data from Kafka topic(s) and send it to a destination. In the diagram below, we can see how it would look if we wanted to take messages from a Kafka topic and store them in Snowflake. Converters Converters are the most confusing component to set up because it's easy to misunderstand and misconfigure them. Since Kafka messages are just bytes, converters translate data between the internal data format used by Kafka Connect and the format specific to the source or sink connector. When we store messages in Kafka topics they will always get serialized, even if we do not specify the serialization format. The Kafka producer will use the String serializer by default to serialize the message and save it to a topic as an array of bytes. Usually, you will use serialization formats like Avro or Protobuf for more complex pipelines. Since we are using both connectors and converters, let's refer to them together as a Kafka Connect system to make it less confusing. We have two systems: a source system and a sink system.  You may see that in some places, those two components are both referred to as connectors, but that is why many people get confused. We will expand both previous diagrams to demonstrate how both source and sink systems look. Source system As you can see in the diagram, the source connector (in our case S3 connector) establishes the connection to S3 and starts polling the desired file. After the S3 connector gets a file, it will hand out the contents of that file to the converter. The converter will take that content and serialize it with the specified serializer to the specified format. After serialization, it will publish the serialized message to a Kafka topic. Sink system In sink systems, the converter first takes the serialized message from a Kafka topic, deserializes it, and hands it over to the sink connector. After that, the sink connector (in our case the Snowflake connector) saves the formatted message into the Snowflake table. If you are still confused, imagine that the connector connects Kafka Connect to a source/destination and acts as a bridge between those two points while the converter does all the data transformations (serialization/deserialization). Understanding converters is a crucial part of setting up Kafka Connect. I highly recommend spending some time to understand connectors and converters before attempting to set up Kafka Connect for a complex pipeline. It will save you a lot of time in the long run.  Confluent published a great blog about converters here: Kafka Connect Deep Dive – Converters and Serialization Explained | Confluent.  Dead Letter Queue (DLQ) A Dead letter queue or DLQ is a component that can be easily overlooked since it is not a required component for setting up Kafka Connect. In my opinion, DLQ is a must for any production grade data pipeline. DLQ is a component that is responsible for handling messages that failed to be consumed. In terms of Kafka, DLQ is a separate topic that is only used to catch messages that had some issues and could not be consumed. The most common issue when this happens is when using the wrong converter. If you have worked with Avro, there is a good chance you saw the “Unknown magic byte!” error. That error occurs when you try to use an Avro converter to deserialize messages that are not Avro. A wrong converter is one example, but there can be more issues. To avoid losing messages, we need a DLQ that will handle those cases. By default, the Kafka Connect cluster will stop if it cannot deserialize a message, and we will need to manually fix the issue and restart it.  In production environments we cannot allow downtime nor lose messages and that's where DLQ can help us. We can configure Kafka Connect to not crash when a message cannot be consumed or deserialized but instead send it to the DLQ topic. These are the properties that we need to define if we want to do that: "errors.tolerance": "all", "errors.deadletterqueue.topic.name": "MY_DLQ_TOPIC", "errors.deadletterqueue.topic.replication.factor": 3, "errors.deadletterqueue.context.headers.enable": true Schema registry Schema registry is an additional service that helps us improve the data quality of our Kafka data. It's a centralized repository where we can store our schemas for our Kafka messages. You can think of schemas as blueprints for how the message should look like. Schema definition contains all the fields and their types that a message needs to have. When a producer wants to publish a message to a topic, it must first validate whether the message satisfies the schema definition. Only then can the message be published. The same applies for consumers. This helps us have standards, and establishes better collaboration across different teams, and helps prevent mistakes and inconsistencies. This is an example how a schema definition looks like: {   "doc": "Sample schema",   "fields": [     {       "doc": "ID of the user. Type Int.",       "name": "id",       "type": "int"     },     {       "doc": "Users full name. Type String.",       "name": "name",       "type": "string"     }   ],   "name": "sampleRecord",   "namespace": "com.dk.mynamespace",   "type": "record" } Kafka Connect is made by Confluent and has many integrations with the Confluent Platform, and also different licenses. To reduce the complexity of the blog, in part 2 we will use the Confluent Platform schema registry.  In case you cannot use the Confluent Platform, Confluent provides the schema registry under Community License and in 99% of the cases you will be fine to host your own instance of the schema registry. The only exception is if you want to offer it as a service and compete with Confluent. In those cases, feel free to use alternatives like Karapace.  Conclusion When it comes to streaming messages from Kafka to Snowflake, Kafka Connect was a clear winner for us. It provided us with high reliability, scalability and fault tolerance while also having out of the box integration with schema registry and dead letter queues. In this blog, we learned about Kafka Connect, how to use it, and which features are necessary for a production-grade big data pipeline.  For a practical guide on how to set up Kafka Connect; stay tuned for part 2 of the blog where we configure and deploy three Kafka Connect clusters (avro, protobuf and string) on Kubernetes and demonstrate the process how we can utilize Schema Registry and DLQ (Dead Letter Queue) to establish a production grade data pipeline. The architecture we will set up is shown here:

Keeping secrets secure with Vault inside a Kubernetes cluster In today's world where data plays a huge part in our lives, it is important to keep that data safe and secure. Everyday sites are getting hacked, databases breached and personal data stolen. That can lead to huge financial losses but can also damage the reputation of the company. According to CSOonline about 3.5 billion people saw their personal data stolen in the top two biggest breaches of this century alone. As data grows so does the need for storage where it will be stored. To handle more storage we need more processing power (servers) that will manage those databases. The more servers we involve in this process the more we increase the risk that those servers will get compromised and our database credentials stolen. To prevent that, engineers are working on tools that can help us minimize that risk or at least reduce the damage inflicted. (more…)