Tutorial: Get started with Named Channels using the SDK

This tutorial provides step-by-step instructions for setting up and running a Named Channel demo application with the snowpipe-streaming SDK. Named Channels provide ordered, exactly-once ingestion. Exactly-once recovery requires records retained in the source or durable application-managed storage for replay.

For Elastic Channels (the recommended starting point for most new applications), see Tutorial: Get started with Elastic Channels (SDK).

Prerequisites

Before you run the demo, ensure that you meet the following prerequisites:

  • Snowflake account: Verify that you have access to a Snowflake account. You will need a user with sufficient privileges (e.g., ACCOUNTADMIN or USERADMIN for the initial setup) to create the dedicated user and custom role detailed in Step 1: Configure Snowflake objects.

  • Network access: Ensure that your network allows outbound connectivity to Snowflake and Amazon S3 or Google Cloud Platform (GCS) or Azure Blob Storage. Adjust firewall rules if necessary because the SDK makes REST API calls to Snowflake and to your cloud storage provider.

    • To verify network connectivity, use the following command:
    # Test connectivity to Snowflake; replace with your account URL
curl -I https://<your_account_identifier>.snowflakecomputing.com

# Test connectivity to AWS S3
curl -I https://s3.amazonaws.com

# Test connectivity to GCS
curl -I https://storage.googleapis.com

# Test connectivity to Azure Blob Storage
curl -I https://azure.blob.core.windows.net  or curl -I https://<your_account_name>.blob.core.windows.net

  • Java Development Environment: Install Java 11 or later, and Maven for dependency management.

  • Python: Install Python version 3.9 or later.

  • Node.js: Install Node.js version 20 or later.

  • System requirements: The SDK requires glibc version 2.26 or later. You can check your current glibc version with:

    ldd --version

  • Snowpipe Streaming SDKs and the sample code:

    Download the sample code for your preferred language from the Snowpipe Streaming SDK examples in the GitHub repository.

Prerequisites and setup

The following steps configure the Snowflake objects, authentication, and SDK dependency required for Named Channel ingestion.

Step 1: Configure Snowflake objects

Before you can use the snowpipe-streaming SDK, you must create a target table within your Snowflake environment. Unlike the classic architecture, the high-performance architecture requires a PIPE object for data ingestion. This tutorial uses the default pipe that is automatically created at ingest time for your target table. If you require additional features, such as in-flight transformations or clustering at ingest time, see CREATE PIPE.

Generate a key pair for authentication

Generate a private-public key pair for authentication using OpenSSL. For more information, see Key-pair authentication and key-pair rotation.

Run the following commands in your terminal to generate the keys:

openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt
openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

PUBK=$(cat ./rsa_key.pub | grep -v KEY- | tr -d '\012')
echo "ALTER USER MY_USER SET RSA_PUBLIC_KEY='$PUBK';"

Important

Save the generated rsa_key.p8 (private key) and rsa_key.pub (public key) files securely. You will use these keys in subsequent authentication steps.

Create database, schema, table, and configure user authentication

Run the following SQL commands in your Snowflake account; for example, by using Snowsight or Snowflake CLI). You must have a role with permissions to create users, roles, and databases — such as ACCOUNTADMIN or USERADMIN for the first few lines, and then switching to the new role. Replace placeholders like MY_USER, MY_ROLE, MY_DATABASE, and so on, with the names that you want.

-- 1. Create a dedicated role and user (Run with a highly-privileged role)
CREATE OR REPLACE USER MY_USER;
CREATE ROLE IF NOT EXISTS MY_ROLE;
GRANT ROLE MY_ROLE TO USER MY_USER;

-- 2. Set the public key for key-pair authentication
-- NOTE: Replace 'YOUR_FORMATTED_PUBLIC_KEY' with the output of the PUBK variable from the key generation step.
ALTER USER MY_USER SET RSA_PUBLIC_KEY='YOUR_FORMATTED_PUBLIC_KEY';

-- 3. Set the default role (Recommended)
ALTER USER MY_USER SET DEFAULT_ROLE=MY_ROLE;

-- 4. Switch to the new role and create objects
USE ROLE MY_ROLE;
-- NOTE: You may also need to run USE WAREHOUSE YOUR_WH; here if a default warehouse isn't set.

-- Create database and schema
CREATE OR REPLACE DATABASE MY_DATABASE;
CREATE OR REPLACE SCHEMA MY_SCHEMA;

-- Create a target table
CREATE OR REPLACE TABLE MY_TABLE (
    data VARIANT,
    c1 NUMBER,
    c2 STRING
);

-- 5. Configure authentication policy (Optional, but recommended for explicit control)
CREATE OR REPLACE AUTHENTICATION POLICY testing_auth_policy
  AUTHENTICATION_METHODS = ('KEYPAIR')
  CLIENT_TYPES = ('DRIVERS');

-- Apply authentication policy (if created)
ALTER USER MY_USER SET AUTHENTICATION POLICY testing_auth_policy;

Note

The data column in the sample table is a VARIANT type. The high-performance SDK requires that data for this column be passed as a native object; for example, a Java Map, Python dictionary, or JavaScript object. Passing a raw JSON string results in the data being stored as a string literal.

Step 2: Configure an authentication profile

The demo application requires a profile.json file to store connection settings, including authentication details. The SDK uses key-pair authentication for secure connections.

Create a profile configuration file

Create or update the profile.json file in the root directory of your demo project.

profile.json template

{
    "user": "MY_USER",
    "account": "your_account_identifier",
    "url": "https://your_account_identifier.snowflakecomputing.com:443",
    "private_key_file": "rsa_key.p8",
    "role": "MY_ROLE"
}

Replace the placeholders:

Step 3: Set up the demo project

Download: Sample Java code

Add the JAR dependency

To include the Snowpipe Streaming SDK, add the following dependency to your Maven pom.xml. Maven automatically downloads the JAR from the public repository.

<dependency>
    <groupId>com.snowflake</groupId>
    <artifactId>snowpipe-streaming</artifactId>
    <version>YOUR_SDK_VERSION</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.18.1</version>
</dependency>

Important

Replace YOUR_SDK_VERSION with the specific version available on Maven Central.

Place the profile file

Ensure that the profile.json file that you configured in Step 2: Configure an authentication profile is located in the root directory of your project.

Open and use a Named Channel

Use a Named Channel when your application requires ordering within a channel or exactly-once recovery. Complete Steps 1 through 3 in Prerequisites and setup before running the following code.

The examples open MY_CHANNEL with an initial offset token, asynchronously append three rows with offset tokens 1, 2, and 3 without waiting on each individual append, retrieve channel status, then wait once for offset token 3 to commit before closing.

Append rows as they arrive; the SDK combines appends internally using time and size thresholds. Submit rows serially in source order within each Named Channel, but don’t wait for every row to commit. Bound outstanding work and checkpoint periodically without collecting rows into another batch before submitting them.

ObjectMapper mapper = new ObjectMapper();
JsonNode profile = mapper.readTree(Files.readAllBytes(Paths.get("profile.json")));
Properties properties = new Properties();
profile.fields().forEachRemaining(
    entry -> properties.put(entry.getKey(), entry.getValue().asText()));

try (SnowflakeStreamingIngestClient client =
    SnowflakeStreamingIngestClientFactory.tableBuilder(
            "demo-client", "MY_DATABASE", "MY_SCHEMA", "MY_TABLE")
        .setProperties(properties)
        .build()) {
  SnowflakeStreamingIngestChannel channel =
      client.openChannel("MY_CHANNEL", "0").getChannel();

  Map<String, Object> row1 = Map.of("DATA", Map.of("event_id", 1, "status", "active"), "C1", 1, "C2", "example");
  Map<String, Object> row2 = Map.of("DATA", Map.of("event_id", 2, "status", "active"), "C1", 2, "C2", "example");
  Map<String, Object> row3 = Map.of("DATA", Map.of("event_id", 3, "status", "active"), "C1", 3, "C2", "example");

  channel.appendRow(row1, "1");
  channel.appendRow(row2, "2");
  channel.appendRow(row3, "3");

  ChannelStatus status = channel.getChannelStatus();

  // Wait until offset 3 or a later offset is committed.
  channel.waitForCommit(
      token -> token != null && Long.parseLong(token) >= 3,
      Duration.ofMinutes(1)).get();
  channel.close();
}

Important

Don’t call waitForCommit after every row or small batch. Waiting on every append serializes ingestion around commit latency and defeats the purpose of asynchronous appends. waitForCommit polls channel status until the target offset commits, so reserve it for checkpoints, source handoffs, or graceful shutdown.

See the full Java, Python, and Node.js samples for complete examples.

Run the application, then query the target table to verify the rows. In a production application, retrieve the latest committed offset when you open the Named Channel and resume the source from the next record. For more information, see Offset tokens and exactly-once delivery.