Streaming Data from Kafka to Snowflake: 2 Easy Methods

In this digital world, everything is available and controlled over the internet. This process generates a lot of data and it is difficult to perform real-time analysis, formulate insights, detect frauds and anomalies, notify users, etc.

In this blog post, we aim to discuss an event streaming platform – Kafka, and a cloud-hosted data warehouse – Snowflake. We will also cover how you can connect Kafka to Snowflake to leverage Snowflake’s data analytics capability and generate insights from the data.

What is Kafka?

Apache Kafka is an open-source distributed system to publish and subscribes to a high volume of messages from one end to another. Kafka uses the Broker’s concept to replicate and persist the message in a fault-tolerant way and segregate the message according to subjects. Kafka is used for building real-time streaming data pipelines and streaming applications to transform data and stream it from its source to its target.

Key Features of Kafka

1. Scalability: Kafka has exceptional scalability and can be scaled easily without downtime.
2. Data Transformation: Kafka offers KStream and KSQL (in the case of Confluent Kafka) for on-the-fly data transformation.
3. Fault-Tolerant: Kafka uses Brokers to replicate and persists the data to make it a fault-tolerant system.
4. Security: Kafka can be combined with available security measures like Kerberos to stream the data securely.
5. Performance: Kafka is distributed and partitioned and has a very high throughput for publishing and subscribing to the message.

What is Snowflake?

Snowflake is a cloud-based, fully managed data warehouse system that provides exceptional data analytics capability. Snowflake uses AWS, Azure, and GCP to manage its cloud infrastructure and data. Snowflake uses SQL to perform queries on the data to transform it and generate insights.

Key features of Snowflake

Managed Platform: Snowflake is a software-as-a-Service cloud-based model that manages hardware and software on its own.

Centralized Repository: Snowflake uses the cloud storage system to store data centrally and is accessible across all compute nodes in the data warehouse.

High-end Performance: Each Snowflake cluster is associated with MPP (Massive Parallel Processing) to process the load parallelly and individually without affecting other nodes’ performance.

Security: Snowflake provides exceptional protection for data at rest and on the flight to protect user information and business factors.

No Hardware: With Snowflake, there is no requirement to set up, configure, install and manage hardware.

Scalable: Snowflake can easily scale up and down based on the volume of data.

Fault-Tolerant: Snowflake provides exceptional fault-tolerant capabilities to recover accidentally dropped snowflake objects.

Standard and Extended SQL Support: Snowflake has the best support for ANSI SQL along with advanced SQL functionality like merge, lateral view, statistical functions, and many more.

What are the Methods to connect Kafka to Snowflake?

Here are the methods you can use to connect Kafka to Snowflake in a seamless fashion:

• Method 1: Using Apache Kafka to connect Kafka to Snowflake
• Method 2: Using Hevo to connect Kafka to Snowflake

Method 1: Using Apache Kafka to connect Kafka to Snowflake

Pre-requisites:

• Snowflake account with Read/Writes access to the database, schema, and tables.
• Up and running Confluent Kafka or Apache Kafka. 
• Kafka Connectors (Apache Kafka or Confluent). See details officially.
• Working knowledge of Snowflake tables and data.
• Working knowledge of Kafka.

Step 1: Kafka Installation

Snowflake provides a connector to Apache Kafka and Confluent Kafka. To download and install Kafka on standalone or distributed mode, you might want to look at the official documentation – 

• Apache Kafka Installation Guide
• Confluent Kafka Installation Guide

Step 2: Download and Installation of Connector

Once you successfully installed Kafka, you need to download the Kafka-Snowflake connector to extract the streams of data from Kafka topics.

You can download the Kafka Connector from here :

• Confluent Hub: https://www.confluent.io/hub/
• Maven Repository [Apache Kafka]: https://mvnrepository.com/artifact/com.snowflake

After you download the connector, you need to install them.

• To install the connector for Confluent Kafka, refer to this guide – https://docs.confluent.io/current/connect/userguide.html
• To Install the connector on Apache Kafka, place the downloaded jar file in the <kafka_dir>/libs folder.

Step 3: Create Database and Schema on Snowflake

To stream the data from Kafka to Snowflake; first, you need to create the database and schema as these are the mandatory parameters required in the configuration parameters of Kafka Connectors. To create, head out to Snowflake’s query panel and execute the following command.

create schema kafka_schema;
create database kafka_db;

Step 4: Creating a Role on Snowflake to use Kafka Connector

In Snowflake, it is recommended to create a separate role and user to access Kafka so that the access can be revoked if needed. The current user should have the privileges to create a new role and grant privileges. Execute the below steps on the Snowflake query panel to create new roles and grant necessary rights.

-- Use a role that can create and manage roles and privileges:
use role securityadmin;
-- Create a Snowflake role
create role kafka_role_1;
-- Grant privileges on the database:
grant usage on database kafka_db to role kafka_role_1;
-- Grant privileges on the schema:
grant usage on schema kafka_schema to role kafka_role_1;
grant create table on schema kafka_schema to role kafka_role_1;
grant create stage on schema kafka_schema to role kafka_role_1;
grant create pipe on schema kafka_schema to role kafka_role_1;
-- Grant the custom role to an existing user:
grant role kafka_role_1 to user kafka_user_1;

Step 5: Kafka Connector Configuration

Now that you have installed Kafka, created the database, and schema, and assigned the roles in Snowflake, you need to configure the Kafka Connector based on Kafka’s installation type, i.e., in Standalone mode or Distributed mode.

Let’s see how you can add the parameters for the distributed and standalone modes. The next section explains the usage of each parameter in detail.

Distributed Mode

Create a configuration file called connect-distributed.properties at the location <kafka_dir>/config/. The sample configuration is shown below:

{
  "name":"CustomerData",
  "config":{
"connector.class":"com.snowflake.kafka.connector.SnowflakeSinkConnector",
     "tasks.max":"8",
     "topics":"dev.customerData,dev.orderData",
"Snowflake.topic2table.map": "topic1:cutomer_data,topic2:orders_data",
     "buffer.count.records":"10000",
    "buffer.flush.time":"60",
     "buffer.size.bytes":"5000000",
"snowflake.url.name":"myaccount.us-west-2.snowflakecomputing.com:443",
    "snowflake.user.name":"user-name",
    "snowflake.private.key":"password",
    "snowflake.private.key.passphrase":"adgshf82345tbgfv",
    "snowflake.database.name":"kafka_db",
    "snowflake.schema.name":"kafka_schema",
"key.converter":"org.apache.kafka.connect.storage.StringConverter",
"value.converter":"com.snowflake.kafka.connector.records.SnowflakeAvroConverter", "value.converter.schema.registry.url":"http://localhost:8081",
    "value.converter.basic.auth.credentials.source":"USER_INFO",
"value.converter.basic.auth.user.info":"user-name:MyStrongPassword"
  }
}

Standalone Mode

For Standalone mode create a configuration file, at <kafka_dir>/config/connect-standalone.properties and populate the file with all connector configuration information.

Sample file – 

connector.class=com.snowflake.kafka.connector.SnowflakeSinkConnector
tasks.max=8
topics=dev.customerData,dev.orderData
snowflake.topic2table.map= topic1:cutomer_data,topic2:orders_data
buffer.count.records=10000
buffer.flush.time=60
buffer.size.bytes=5000000
snowflake.url.name=myaccount.us-west-2.snowflakecomputing.com:443
snowflake.user.name=user-name
snowflake.private.key=password
snowflake.private.key.passphrase=jkladu098jfd089adsq4r
snowflake.database.name=kafka_db
snowflake.schema.name=kafka_schema
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=com.snowflake.kafka.connector.records.SnowflakeAvroConverter
value.converter.schema.registry.url=http://url/to/schema/registry:8081
value.converter.basic.auth.credentials.source=USER_INFO
value.converter.basic.auth.user.info=jane.smith:MyStrongPassword

Step 6: Kafka Configuration Properties

Let’s have a look at the description and usage of the property defined above:

Required Properties

name – A unique application name across the Kafka connectors.

Connector.class – Class name of the connector, in this case it will be – com.snowflake.kafka.connector.SnowflakeSinkConnector.

topics – Comma-separated list of topics.

snowflake.url.name – Snowflake URL to access Snowflake account.

snowflake.user.name – User login name for the Snowflake account.

snowflake.private.key – The private key to authenticate the user. For more details on the Snowflake’s private key, follow the Topic – Key Pair Authentication.

snowflake .database.name – The name of the database that contains the table to insert rows.

snowflake.schema.name – The name of the schema that contains the database and table to insert rows.

header.converter – This is required when the Kafka messages are in Avro and contains a header. 

key.converter – This is the Kafka record’s key converter (e.g. “org.apache.kafka.connect.storage.StringConverter”).

value.converter – If the records are formatted in JSON, this should be “com.snowflake.kafka.connector.records.SnowflakeJsonConverter”.

Optional Properties

Snowflake.private.key.passphrase – Phrase to decrypt the private key.

tasks.max – Number of tasks (same as CPU cores).

snowflake.topic2table.map – If the topic name and table name are different then use this parameter to map the topic against the table. By default, Snowflake assumes the table name to be the same as the topic name.

buffer.count.records – Buffered records in Kafka before ingestion to Snowflake. The default value is 10000 records.

buffer.size.bytes – Size in bytes of records buffered in memory before ingestion to Snowflake. The default value for this is 5000000 (5 MB).

For the complete list of configuration parameters, refer to Kafka Configuration Parameters.

Step 7: Starting Kafka and Kafka Connector

Once you configured the parameters correctly, you need to start the Kafka Connector to pull data from the topic and load it to Snowflake.

Starting Kafka

1. To start the Confluent Kafka, refer to the Confluent Official documentation.
2. To download and configure Apache Kafka, refer to the Apache Kafka Official documentation.

Starting Kafka Connector

To start the Kafka Connector, use the given commands for distributed mode or standalone mode.

1. Distributed Mode:

curl -X POST -H "Content-Type: application/json" --data @<config_file>.json http://localhost:8083/connectors

2. Standalone Mode:

<kafka_dir>/bin/connect-standalone.sh <kafka_dir>/<path>/connect-standalone.properties <kafka_dir>/config/connect-standalone.properties

Step 8: Testing and Using Kafka Connector

1. Check whether Kafka and the Kafka Connector are running.
2. Check whether the appropriate topic has been created or not. For Confluent Kafka, you can check this using the Confluent UI.
3. Create a publishing mechanism to publish the data into Snowflake or use an existing one.
4. Ensure that the messages are in the right format (JSON or Avro).
5. Create a configuration file, as described above.
6. Kafka automatically creates the table with proper schema in Snowflake if it is not there. Hence, it is recommended to let the connector generate the table for you to minimize the possibility of a schema mismatch.
7. Grant the privileges as described above.
8. Start publishing the message to the specified Kafka topic.
9. After a few moments, the data will start flowing to the Snowflake table subscribed to the topic. Check the Snowflake table to see the records inserted by the Kafka Connector.

Limitations of Method 1:

The Kafka Connector is used for Kafka to Snowflake connections it has its own set of drawbacks.

• this method requires higher technical expertise to set up the connections.
• Requires installation of all the components into your system taking space, effort, and time.
• It requires a lot of configuration components which performing manually is a hectic task and more prone to errors.
• It requires manual testing of the connection and if an error occurs the process needs to be repeated.

The limitations reduce the overall efficiency of the connector and there is no support provided. But, these limitations can be overcome very easily by using the second method.

Method 2: Using Hevo to connect Kafka to Snowflake

Hevo Data, an Automated No-code Data Pipeline, can help you connect Kafka to Snowflake. It supports 150+ Data Sources (including 40+ free sources) like Kafka and stores the data straight into your Data Warehouse like Snowflake. With Hevo in place, you can automate the Kafka to Snowflake Connection which will reduce the effort and errors immensely. You can connect to any data source of your choice and leverage Hevo’s blazing-fast Data Pipelines to help you Extract, Add Value to your Raw Data, and Aggregate this Granular Data to build your Single Source of Truth in a matter of minutes! All this without writing any code!

Hevo offers an entirely Automated ETL offering to deliver your data in real time without any loss from source to destination.

With Hevo’s point-and-click interface, loading data from Kafka to Snowflake comes down to 2 simple steps:

• Step 1: Connect Hevo to Apache Kafka by entering the Pipeline Name, and Bootstrap Server(s), Use SSL, and Ingest Data From.

• Step 2: Complete Kafka Snowflake migration by providing your destination name, account name, region of your account, database username and password, database and schema name, and the Data Warehouse name.

Here are more reasons to love Hevo:

1. Secure: Hevo has a fault-tolerant architecture that ensures that the data is handled in a secure, consistent manner with zero data loss.
2. Auto Schema Mapping: Hevo takes away the tedious task of schema management & automatically detects the schema of incoming data and maps it to the destination schema.
3. Minimal Learning: Hevo, with its simple and interactive UI, is extremely simple for new customers to work on and perform operations.
4. Hevo is Built to Scale: As the number of sources and the volume of your data grows, Hevo scales horizontally, handling millions of records per minute with very little latency.
5. Incremental Data Load: Hevo allows the transfer of data that has been modified in real-time. This ensures efficient utilization of bandwidth on both ends.
6. Live Support: The Hevo team is available round the clock to extend exceptional support to its customers through chat, email, and support call
7. Live Monitoring: Hevo allows you to monitor the data flow and check where your data is at a particular point in time.

With continuous Real-Time data movement, Hevo allows you to combine Kafka data along with your other data sources and seamlessly load it to Snowflake with a no-code, easy-to-setup interface. Try our 14-day full-feature access free trial!

Conclusion

In this article you have looked at how you can ingest streaming data from Kafka to Snowflake, the data warehouse, to perform analytics. Also, check out the Kafka MongoDB connection.

If you are looking for a fully managed solution that performs Automated Kafka to Snowflake Connection then Hevo Data is the right choice for you!

Connect Kafka to Snowflake without writing any code

Hevo will help you perform ETL and automated data integration processes of both the Data Sources and the Data Destinations. It provides a consistent & reliable solution to Connect Kafka or any other data Source to Snowflake in real time and always has analysis-ready data

Want to take Hevo for a spin? Sign up for a 14-day free trial and see the difference yourself!