Copy page

Outputs

Snowflake Snowpipe Streaming

Stream records into Snowflake in near real time via the Snowpipe Streaming high-performance architecture. Records become queryable in the target table within seconds of being written, with no intermediate files and no warehouse required for ingest.

Prerequisites

• A Snowflake account on a supported region (high-performance architecture is GA on AWS, Azure, and GCP commercial regions).
• A Snowflake user dedicated to Monad ingest, with key-pair auth enabled.
• A role with USAGE on the database/schema, INSERT on the target table, and OPERATE/MONITOR on the pipe.
• A target table with columns matching the JSON keys you stream.
• A STREAMING pipe (CREATE PIPE ... DATA_SOURCE = (TYPE = 'STREAMING')) pointing at that table.
• The Monad node must be able to reach *.snowflakecomputing.com over HTTPS.

Setup Instructions

Step 1: Generate an RSA key pair

Generate an unencrypted PKCS#8 RSA key pair locally:

Code
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out monad_key.pem -nocrypt
openssl rsa -in monad_key.pem -pubout -out monad_key.pub

Keep monad_key.pem private — you'll paste it into Monad as a secret. The contents of monad_key.pub get registered on the Snowflake user.

See Key-pair authentication and key-pair rotation for details, including rotation procedure.

Step 2: Create the Snowflake user and role

Run these as a role with SECURITYADMIN or higher:

Code
CREATE ROLE monad_streaming_role;

CREATE USER monad_streaming_user
  TYPE = SERVICE
  DEFAULT_ROLE = monad_streaming_role
  RSA_PUBLIC_KEY = 'MIIBIj...your public key body, no header/footer/newlines...';

GRANT ROLE monad_streaming_role TO USER monad_streaming_user;

The connector signs JWTs that do not carry a role claim. The user's DEFAULT_ROLE is what Snowflake uses to authorize the request, so it must point at the role that owns the grants below.

Step 3: Create the target table

The table schema must match the JSON shape you're streaming. For semi-structured ingest, a single VARIANT column works well:

Code
USE ROLE sysadmin;
CREATE DATABASE IF NOT EXISTS monad_db;
CREATE SCHEMA   IF NOT EXISTS monad_db.monad_schema;

CREATE TABLE monad_db.monad_schema.cloudtrail_events (
  record VARIANT
);

For typed columns, define them explicitly and ensure your upstream pipeline emits matching keys.

Step 4: Create the STREAMING pipe

The pipe is the ingest target the connector writes to. It must be created with DATA_SOURCE = (TYPE = 'STREAMING'):

Code
CREATE PIPE monad_db.monad_schema.cloudtrail_pipe
  AS COPY INTO monad_db.monad_schema.cloudtrail_events
  FROM TABLE (
    DATA_SOURCE => (TYPE => 'STREAMING')
  )
  MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

For a typed schema, replace MATCH_BY_COLUMN_NAME with an explicit column list. See CREATE PIPE.

Step 5: Grant the role access

Code
USE ROLE sysadmin;

GRANT USAGE   ON DATABASE monad_db                                TO ROLE monad_streaming_role;
GRANT USAGE   ON SCHEMA   monad_db.monad_schema                   TO ROLE monad_streaming_role;
GRANT INSERT  ON TABLE    monad_db.monad_schema.cloudtrail_events TO ROLE monad_streaming_role;
GRANT OPERATE, MONITOR ON PIPE monad_db.monad_schema.cloudtrail_pipe TO ROLE monad_streaming_role;

Step 6: Configure the output in Monad

In the pipeline editor, add the Snowflake Snowpipe Streaming output and fill in the settings below.

Configuration

Settings

Setting	Type	Required	Description
Account	string	Yes	Snowflake account identifier, in the form orgname-account_name (e.g. acme-prod1). This is the same identifier you use in the Snowflake URL.
User	string	Yes	The Snowflake user for key-pair auth. Its DEFAULT_ROLE must be a role with access to the pipe.
Database	string	Yes	Database containing the pipe. Alphanumeric and underscores only.
Schema	string	Yes	Schema within the database that contains the pipe. Alphanumeric and underscores only.
Pipe	string	Yes	Name of the pre-existing STREAMING pipe. Alphanumeric and underscores only.
Channel Prefix	string	No	Prefix used for channel names. Final channel names are {prefix}_{uuid}_{i}, where uuid is a fresh random identifier. And i ranges over the channel pool. Defaults to monad.
Private Key	secret	Yes	RSA private key (PEM) for the user. Either raw PEM or base64-encoded PEM is accepted.
Batch Config	object	No	Controls when a batch is flushed to Snowflake. See below.

Batch tuning

Field	Default	Min	Max	Notes
Records per batch	10,000	1	50,000	Soft cap; flush also fires on size/time.
Batch size (bytes)	2 MiB	1 MiB	3.5 MiB	Snowflake caps each appendRows body at 4 MiB; we stay below that to leave headroom for JSON framing.
Publish rate (seconds)	10	1	60	Maximum time a partial batch waits before being flushed.

The connector picks the parallel channel count from the configured batch size: smaller batches get more parallel channels (up to 32) to keep per-second throughput up; larger batches use fewer channels since each one already moves more bytes per request. You don't tune this directly — choose the batch size that matches your freshness/throughput goals and the channel pool sizes itself.

Debugging

Status checks in Snowflake

Do not use SYSTEM$PIPE_STATUS — that's the classic Snowpipe function and its fields (pendingFileCount etc.) don't apply to streaming pipes.

For the high-performance architecture, use:

Code
-- List all open channels on the pipe and their commit progress.
-- This is the most useful diagnostic: client_sequencer / row_sequencer /
-- last_committed_offset_token / last_inserted_timestamp tell you whether
-- each channel is advancing under load.
SHOW CHANNELS IN PIPE MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_PIPE;

-- Pipe definition and current status
DESCRIBE PIPE MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_PIPE;

-- Per-pipe ingest volume and credit usage (real-time, INFORMATION_SCHEMA)
SELECT *
FROM TABLE(INFORMATION_SCHEMA.PIPE_USAGE_HISTORY(
  DATE_RANGE_START => DATEADD('hour', -1, CURRENT_TIMESTAMP()),
  PIPE_NAME => 'MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_PIPE'
));

-- Same data, longer retention, ~45 min lag (ACCOUNT_USAGE)
SELECT *
FROM SNOWFLAKE.ACCOUNT_USAGE.PIPE_USAGE_HISTORY
WHERE PIPE_NAME = 'MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_PIPE'
ORDER BY START_TIME DESC
LIMIT 100;

COPY_HISTORY does not record successful Snowpipe Streaming row inserts — they commit directly into the table. It only surfaces parse/load errors. An empty COPY_HISTORY result means no errors, not no traffic.

See INFORMATION_SCHEMA functions for the full set.

Confirming rows landed

Code
-- 1. Confirm the column name and type
DESC TABLE MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_EVENTS;

-- 2. Plain row count — works regardless of schema
SELECT COUNT(*) FROM MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_EVENTS;

-- 3. If the target is a single VARIANT column (e.g. named "record"),
-- you can pull the latest event timestamp:
SELECT COUNT(*), MAX(record:eventTime::TIMESTAMP_NTZ)
FROM MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_EVENTS;

If Monad reports append rows ok but the row count isn't climbing, run SHOW CHANNELS IN PIPE … and check last_committed_offset_token — if it's advancing but the table is empty, the schema mismatch is silently dropping rows.

Troubleshooting

Common Issues

1. snowflake 408: stream timeout — Snowflake received the bytes but couldn't commit them inside its server-side ack window. Usually the pipe is congested. Retries are safe (continuation tokens make appends idempotent). If sustained, drop concurrent writers against this pipe or shard to a second pipe.
2. Client.Timeout exceeded while awaiting headers — the HTTP client gave up before Snowflake responded. Check network latency from the Monad cluster to the ingest host; firewalls or NAT gateways are common culprits.
3. JWT token is invalid / 390144 — the public key on the Snowflake user doesn't match the private key in Monad. Re-run DESC USER and verify the RSA_PUBLIC_KEY fingerprint matches.
4. Insufficient privileges to operate on pipe — the user's DEFAULT_ROLE doesn't have OPERATE on the pipe. The JWT carries no role claim, so workspace/session role overrides won't help — fix DEFAULT_ROLE.
5. No rows in the target table despite successful appends — usually a schema mismatch. Check INFORMATION_SCHEMA.COPY_HISTORY for FIRST_ERROR_MESSAGE.
6. continuation token mismatch followed by reopen — expected behavior when a request times out and the connector recovers. If frequent, treat as a symptom of pipe-side congestion.
7. Region mismatch — high-performance architecture is not yet available in every region. Confirm your account's region from the Snowpipe Streaming high-performance overview.

Best Practices

• One pipe per data shape, not one pipe per pipeline. Reuse pipes across pipelines whenever the target table is the same.
• Use semi-structured (VARIANT) columns for schema-evolving sources like Cloudtrail. You can flatten downstream with views or dynamic tables.
• Rotate keys at least annually. Snowflake supports two active public keys per user (RSA_PUBLIC_KEY and RSA_PUBLIC_KEY_2) so you can rotate without downtime — see Key-pair rotation.
• Monitor PIPE_USAGE_HISTORY weekly to track credit consumption and ingestion latency trends.

Cost

Snowpipe Streaming high-performance ingest is billed on a serverless compute model — there is no warehouse to size. Costs vary by region, edition, and ingest volume. The current per-credit and per-feature rates are published in Snowflake's Credit Consumption Table (PDF). That document is the source of truth and is updated by Snowflake periodically; we do not mirror its rates here.

To see your actual usage, query:

Code
SELECT *
FROM SNOWFLAKE.ACCOUNT_USAGE.PIPE_USAGE_HISTORY
WHERE PIPE_NAME = 'MONAD_DB.MONAD_SCHEMA.CLOUDTRAIL_PIPE'
ORDER BY START_TIME DESC
LIMIT 100;

Related Articles

• Snowpipe Streaming high-performance architecture overview
• Snowpipe Streaming high-performance REST API reference
• CREATE PIPE
• Key-pair authentication
• INFORMATION_SCHEMA functions
• Credit Consumption Table (PDF)