Pipe Creation

Part of Streaming Transformations. See all pages in this section.

A transformation is part of the PIPE definition. The simplest transformation is a filter: a WHERE clause keeps only the records that match a predicate, and every other record is dropped before it lands in the target table.

CREATE [ OR REPLACE ] PIPE <pipe_name>
  AS COPY INTO <target_table> ( <col1> [, <col2>, ...] )
  FROM (
    SELECT <select_list>
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
    [ WHERE <filter_predicate> ]
  );

Parameters:

  1. <target_table> — Required name of the destination table. Interactive tables, standard Snowflake tables, and managed Iceberg tables are currently supported. We strongly recommend that the ERROR_LOGGING property be set for the target table. See error logging in Snowpipe Streaming. Information about rows that could not be ingested will be placed there.
  2. <select_list> — Expressions written to the target columns, in the same order as the column list.
  3. <filter_predicate> — Boolean expression applied to stream records. See WHERE clause syntax.

Note

An explicit column list is required in both the COPY INTO target and the SELECT subquery.

Important

Before you start, review Required privileges below for the roles/grants needed. See Monitoring & Troubleshooting for how to check pipe health once it’s running.

Example

The following pipe lands only the log records at severity WARN or above:

CREATE TABLE app_logs (
  event_ts TIMESTAMP_NTZ,
  service  STRING,
  severity STRING,
  message  STRING
);

CREATE OR REPLACE PIPE app_logs_pipe
  AS COPY INTO app_logs (event_ts, service, severity, message)
  FROM (
    SELECT
      $1:ts::TIMESTAMP_NTZ,
      $1:service::STRING,
      $1:severity::STRING,
      $1:message::STRING
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
    WHERE $1:severity::STRING IN ('WARN', 'ERROR')
  );

Required privileges

See Usage notes of CREATE PIPE for required privileges.

Limitations

  1. Sources. Streaming Transformations currently supports only Snowpipe Streaming with high-performance architecture. This includes Kafka Connector v4 and Openflow.
  2. Sinks. Sink targets are currently limited to standard Snowflake tables, managed Iceberg tables, and Interactive tables.
  3. Lifecycle. The definition of a PIPE cannot be modified via ALTER PIPE. To add, remove, or change a transformation, drop the pipe and create a new one. Review Recreating pipes for related considerations.
  4. Pre-clustering. Pre-clustering is currently not supported.

See also

To enrich records from a dimension table, expand arrays, call scalar UDFs, or keep materialized views of the landed rows, see the pages that follow. The Scenarios show real-world use cases that combine several transformations. For the complete syntax with all features combined, see the Reference.

Next: Stream filtering