Snowflake's Snowpipe streaming capabilities are designed for rowsets with variable arrival frequency. It focuses on lower latency and cost for smaller data sets. This helps data workers stream rows into Snowflake without requiring files with a more attractive cost/latency profile.

Here are some of the use cases that can benefit from Snowpipe streaming:

In our demo, we will use real-time commercial flight data over the San Francisco Bay Area from the Opensky Network to illustrate the solution using Snowflake's Snowpipe streaming and MSK (Amazon Managed Streaming for Apache Kafka).

Note that you can either stream the data into a regular Snowflake table or a Snowflake managed Apache Iceberg table depending on your use case.

The architecture diagram below shows the deployment. An MSK cluster and a Linux EC2 instance (jumphost) will be provisioned in private subnets of an AWS VPC. The Linux jumphost will host the Kafka producer and the Snowflake High Performance (HP) Kafka Connector (v4.x) via Kafka Connect. The HP connector uses a server-side architecture with a PIPE object in Snowflake that manages data processing and buffering, delivering up to 10 GB/s throughput per table with 5-10 second latency.

The Kafka producer calls the data sources' REST API and receives time-series data in JSON format. This data is then ingested into the Kafka cluster before being picked up by the Kafka connector and delivered to a Snowflake table. The data in Snowflake table can be visualized in real-time with AMG (Amazon Managed Grafana) and Streamlit The historical data can also be analyzed by BI tools like Amazon Quicksight.

To participate in the virtual hands-on lab, attendees need the following resources.

The MSK cluster is created in a VPC managed by Amazon. We will deploy our Kafka clients in our own VPC and use security groups to ensure the communications between the MSK cluster and clients are secure.

First, click here to launch a provisioned MSK cluster. Note the default AWS region is us-west-2 (Oregon), feel free to select a region you prefer to deploy the environment.

Click Next at the Create stack page. Set the Stack name or modify the default value to customize it to your identity. Leave the default Kafka version as is. For Subnet1 and Subnet2, in the drop-down menu, pick two different subnets respectively, they can be either public or private subnets depending on the network layout of your VPC. Please note that if you plan to use Amazon MSK Connect later, you should choose private subnets here. For MSKSecurityGroupId, we recommend using the default security group in your VPC, if you do not have the default security group, create one on your own before moving forward. Leave TLSMutualAuthentication as false and the jumphost instance type and AMI id as default before clicking Next.

See below sample screen capture for reference.

Leave everything as default in the Configure stack options page and click Next. In the Review page, click Submit.

In about 10-30 minutes depending on your AWS region, the Cloudformation template provisions an MSK cluster with two brokers. It will also provision a Linux EC2 instance in the subnet you selected. We will then use it to run the Kafka connector with Snowpipe streaming SDK and the producer.

In this step we need to connect to the EC2 instance in order to interact with the MSK cluster.

Go to the AWS Systems Manager console in the same region where you setup the MSK cluster, Click Session Manager on the left pane.

Next, we will set the preferred shell as bash.

Go to General preferences section, type in 60 minutes for idle session timeout value.

Further scroll down to Linux shell profile section, and type in /bin/bash before clicking Save button.

Now you should see the EC2 instance created by the Cloudformation template under Target instances. Its name should be <Cloudformation stack name>-jumphost, select it and click Start session.

Create a key pair in AWS Session Manager console by executing the following commands. You will be prompted to give an encryption password, remember this phrase, you will need it later.

See below example screenshot:

Next we will create a public key by running following commands. You will be prompted to type in the phrase you used in above step.

see below example screenshot:

Next we will print out the public key string in a correct format that we can use for Snowflake.

see below example screenshot:

Run the following command to install the Kafka connector and Snowpipe streaming SDK

Note that the version numbers for Kafka, the Snowflake Kafka connector, and the Snowpipe Streaming SDK are dynamic, as new versions are continually published. We are using the version numbers that have been validated to work.

Go to the MSK console and click the newly created MSK cluster, it should have a substring MSKCluster in its name.

Now switch back to the Session Manager window and execute the following command by replacing <broker string> with the copied values.

Now run the following command to add BS as an environment variable so it is recognized across the Linux sessions.

See the following example screen capture.

Run the following commands to generate a configuration file connect-standalone.properties in directory /home/ssm-user/snowpipe-streaming/scripts for the client to authenticate with the Kafka cluster.

A configuration file connect-standalone.properties is created in directory /home/ssm-user/snowpipe-streaming/scripts

Run the following commands to create a security configuration file client.properties for the MSK cluster

A configuration file client.properties is created in directory /home/ssm-user/snowpipe-streaming/scripts

Now we can run the following commands to create a Kafka topic on the MSK cluster to stream our data.

You should see the response Created topic streaming if it is successful.

To describe the topic, run the following commands:

Login to your Snowflake account as a power user with ACCOUNTADMIN role. Then run the following SQL commands in a worksheet to create the dedicated streaming user, role, database, schema, and warehouse that we will use in the lab. The streaming user uses key-pair authentication only (no password).

Next we need to configure the public key for the streaming user to access Snowflake programmatically.

In the Snowflake worksheet, replace < pubKey > with the content of the file /home/ssm-user/pub.Key (see step 4 by clicking on section #2 Create a provisioned Kafka cluster and a Linux jumphost in AWS in the left pane) in the following SQL command and execute.

See below example screenshot:

At this point, the Snowflake setup is complete.

See below example screen capture.

Note that the High Performance (HP) connector (v4.x) will auto-create a default PIPE object named MSK_STREAMING_TBL-STREAMING in the target schema when the connector starts. This is why the GRANT CREATE PIPE was added earlier.

Finally, we are ready to start ingesting data into the Snowflake table.

Go back to the Linux console and execute the following commands to start the Kafka connector.

Start a new Linux session in step 3 by clicking on section #2 Create a provisioned Kafka cluster and a Linux jumphost in AWS in the left pane.

You should see response similar to screen capture below if everything works well.

Note that in the script above, the producer queries a Rest API that provides real-time flight data over the San Francisco Bay Area in JSON format. The API returns a JSON array, so we use jq -c '.[]' to break it into individual JSON objects — one per Kafka message. This is required because the HP connector with schematization enabled expects each message to be a flat JSON object whose keys map to table columns.

The data includes information such as timestamps, icao numbers, flight IDs, destination airport, longitude, latitude, and altitude of the aircraft, etc. The data is ingested into the streaming topic on the MSK cluster and then picked up by the Snowpipe streaming Kafka connector, which delivers it directly into a Snowflake table msk_streaming_db.msk_streaming_schema.msk_streaming_tbl.

Now, switch back to the Snowflake console and switch to the role MSK_STREAMING_RL. The data should have been streamed into a table, ready for further processing.

To verify that data has been streamed into Snowflake, execute the following SQL commands.

You should see the auto-created pipe MSK_STREAMING_TBL-STREAMING with executionState=RUNNING.

Note that, at this point, you should only see one batch of rows in the table, as we have only ingested data once. We will see new rows being added later as we continue to ingest more data.

Now run the following query on the table.

With schematization enabled, the HP connector automatically creates columns from the JSON keys. You should see columns like RECORD_METADATA, ID, ICAO, LAT, LON, ALT, ORIG, DEST, and UTC — no manual flattening needed.

Now execute the following SQL commands to create a convenience view with timestamps and geohashes for visualization.

The SQL commands create a view that converts the epoch timestamp to a proper timestamp, applies time zone conversion, and uses Snowflake's Geohash function to generate geohashes that can be used in time-series visualization tools like Grafana.

Let's query the view flights_vw now.

We can now write a loop to stream the flight data continuously into Snowflake.

Go back to the Linux session and run the following script.

You can now go back to the Snowflake worksheet to run a select count(1) from flights_vw query every 10 seconds to verify that the row counts is indeed increasing.

Iceberg table ingestion requires version 3.0.0 or later of the Kafka connector and is only supported with Snowpipe Streaming. The HP connector (v4.x) natively supports Iceberg tables with the same configuration used for native tables. The only additional steps are creating an external volume, pre-creating the Iceberg table, and updating the topic-to-table mapping.

Before you create an Iceberg table, you must have an external volume. An external volume is a Snowflake object that stores information about your cloud storage locations and identity and access management (IAM) entities (for example, IAM roles). Snowflake uses an external volume to establish a connection with your cloud storage in order to access Iceberg metadata and Parquet data.

To create an external volume in S3, follow the steps mentioned in this document.

Below is an example to create External Volume below: When creating the external volume, be sure to use the ACCOUNTADMIN role.

After the external volume is created, use the ACCOUNTADMIN role to grant usage to the role.

Note that if you are using a regular table to store the streaming data, Snowpipe streaming will automatically create it for you if it doesn't exist already. But for Iceberg table, you will need to pre-create an Iceberg table referencing the external volume you just created. You can specify BASE_LOCATION to instruct Snowflake where to write table data and metadata, or leave empty to write data and metadata to the location specified in the external volume definition.

Sharing an example of the SQL to create External volumne below:

Since the BASE_LOCATION is specified, it will create a folder based on the value in the STORAGE_BASE_URL of the External Volume.

Since the Scehmatization is enabled in the Kafka Connector, followig privileges are set properly on the table.

Modify the two lines in $HOME/snowpipe-streaming/scripts/snowflakeconnectorMSK.properties

Save the properties file. Note that snowflake.enable.schematization=TRUE is already enabled in the HP connector properties, and the HP connector (v4.x) natively supports Iceberg tables without any additional properties.

We now need to create a new topic streaming-iceberg in MSK cluster by running the following command:

Restart the consumer by issuing the following shell command in a new Session Manager console.

Now ingest some data into the newly created topic by running the following command in a new Session Manager console.

Now, switch back to the Snowflake console and switch to the role MSK_STREAMING_RL. The data should have been streamed into a table, ready for further processing.

To verify that data has been streamed into Snowflake, execute the following SQL commands.

You should see that the table contains the keys in json records as column names with values and RECORD_METADATA columns are populated.

So far we have been hosting the Kafka connector for Snowpipe Streaming on the EC2 instance. You can also use Amazon MSK Connect to manage the connector.

*Note that in order to use MSKC, it is suggested to run your MSK cluster in a private subnet with egress to the internet via the NAT gateway for it to work. See this AWS documentation for more information.

Follow this AWS documentation to create a S3 bucket.

On your EC2 session, run the following commands to compress the libraries into a zipped file.

Fill in the s3 path to your uploaded zip file, e.g. s3://my-mskc-bucket/snowpipeStreaming-mskc-plugins.zip Give the custom plugin a name, e.g. my-mskc-plugin, click Create custom plugin.

Leave all other settings as default, further scroll down to Access permissions. In the Choose service role drop-down menu, select the role created by the Cloudformation template in the beginning of this quickstarts. The role name should look something like this <CFT stack name>-MSKConnectRole-<random characters>. Click Next.

At this point, the Kafka connector for Snowpipestreaming has been configured, it is running and managed by MSKC, all you need to do is to run the source connector to ingest live data continuously as shown in Step 3 of Section 6.

When you are done with the demo, to tear down the AWS resources, simply go to the Cloudformation console. Select the Cloudformation template you used to deploy the MSK cluster at the start of the demo, then click the Delete tab. All the resources that were deployed previously, such as EC2 instances, MSK clusters, roles, etc., will be cleaned up.

See example screen capture below.

After the deletion of the MSK cluster, you will also need to delete the Cloudformation template for VPC if you created your own at the very beginning of the lab.

For Snowflake cleanup, execute the following SQL commands.

In this lab, we built a demo to show how to ingest time-series data using Snowpipe streaming and Kafka with low latency. We demonstrated this using a self-managed Kafka connector on an EC2 instance. However, for a production environment, we recommend using Amazon MSK Connect, which offers scalability and resilience through the AWS infrastructure. Alternatively, if you have infrastructure supported by either Amazon EKS or Amazon ECS, you can use them to host your containerized Kafka connectors as well.

Related Resources