Read from Kafka & Write to Snowflake via Spark Databricks

In this article, we will learn how can we Read from Kafka & write that data in Snowflake Table via Databricks using Spark Structured Streaming process.

But before, we start let's understand some of the key concepts.

What is Kafka ?

Kafka is software designed upon the publish/subscribe messaging pattern. Publish/subscribe messaging is where a sender (publisher) sends a message that is not specifically directed to a receiver (subscriber). The publisher classifies the message somehow and the receiver subscribes to receive certain categories of messages. There are other usage patterns for Kafka, but this is the pattern we focus on in this course.

Publisher/subscriber systems typically have a central point where messages are published, called a broker. The broker receives messages from publishers, assigns offsets to them and commits messages to storage.

The Kafka version of a unit of data an array of bytes called a message. A message can also contain a bit of information related to partitioning called a key. In Kafka, messages are categorized into topics.

What is Snowflake ?

Snowflake is a single platform comprised of storage, compute, and services layers that are logically integrated but scale infinitely and independent from one another.

Snowflake is a cloud data warehouse built on top of the public cloud (AWS / Azure / GCP ) infrastructure and is a true SaaS offering. There is no hardware (virtual or physical) for you to select, install, configure, or manage. There is no software for you to install, configure, or manage. All ongoing maintenance, management, and tuning is handled by Snowflake.

The Kafka Server & Schema

The Kafka server is fed by a separate TCP server that reads the Wikipedia edits, in real time, from the various language-specific IRC channels to which Wikipedia posts them. 

That server parses the IRC data, converts the results to JSON, and sends the JSON to a Kafka server, with the edits segregated by language. The various languages are topics.

For example, the Kafka topic "en" corresponds to edits for en.wikipedia.org.

Note: This Kafka Server is already setup by Databricks & provided the access to it so that we can explore. But if you want to explore your own Kakfa, I would suggest you to create a free trail account on Confluent Cloud.

Ofcourse you need to bring some data in it i.e. Twitter data via the Twitter Development Account ( Some work need to be done there :) ).

Below is the link for Confluent.

Luckily, Databricks made our life easy by already providing the full fledged Kafka Server & we will use it to read the data from it , process it in Databricks Notebooks & write it in to Snowflake Tables.

Lets start the Step by Step Process:

Step1: Reading from Kafka Server into Spark Databricks

In this example , the only column we want to keep is value column because thats the column we have the JSON data.

Step2: Defining the Schema for Streaming data.

Note: We can't inferSchema for any streaming data.

Step3: Parsing the Input json which we got from Kafka i.e. value field via the from_json function.

Step4: Filtering records ( Reducing the Volume). Also just making the columns one level up by creating the aliases.

Step5: Now optionally, We can apply any transformation the way we apply when we process the batch data. ( Please try it out )

Step6: Setup credentials for Snowflake.

Step7: Loading Data into Snowflake

Note: We don't have any direct options which can write the data into Snowflake so I have written a small custom function - writeToSnowFlake() which takes the dataframe & epochId. Basically this function will be called for each micro batch.

Step8: Viewing Data in Snowflake

Note: The Streaming job will always be running until we terminate it either via manually or programmatically. We need to decide when or based on what , we need to terminate our streaming job

Below is the way we can stop it via the command.

streamingQuery.stop()

So with step by step process we are able to read the data from Kafka & Write to Snowflake table via Databricks.

The same thing can be repeated to other streaming sources like Amazon Kinesis, Azure Event Hubs etc & the loading also can be repeated to other Cloud Data Warehouses like AWS Redshift, Azure Synapse Analytics.

Please try it out, & let me know in case you need my help in connecting to those sources and targets.

That marked the end of this article. Please like / Share if you have found it useful. Please provide you feedback in the comment sections.

Thanks for reading it. Will talk to you in my next article.