apache/beam · error · IllegalArgumentException

snowPipe is required for streaming writes.

Error message

snowPipe is required for streaming writes.

What it means

SnowflakeWriteSchemaTransformProvider.expand() detects an unbounded input PCollection as a streaming write and requires a snowPipe name, because streaming inserts go through a Snowflake SnowPipe. If the input is streaming and snowPipe is null or empty, it throws this IllegalArgumentException at pipeline construction.

Solutions

1. Create a SnowPipe in Snowflake and set it in the configuration, e.g. snowPipe: MY_DB.MY_SCHEMA.MY_PIPE.
2. If the write is intended to be batch, ensure the input PCollection is bounded (use a bounded source or window/complete the stream).
3. For batch semantics over a stream, materialize the input (e.g. write to files first) then apply the batch write.

Example fix

// before
config: {serverName: ..., database: ..., } // no snowPipe, streaming input
// after
config: {serverName: ..., database: ..., snowPipe: "MY_DB.MY_SCHEMA.MY_PIPE"}

boolean streaming = pcollection.isBounded() == PCollection.IsBounded.UNBOUNDED;
if (streaming && (config.getSnowPipe() == null || config.getSnowPipe().isEmpty())) {
  throw new IllegalArgumentException("Streaming input requires snowPipe to be set");
}

try {
  return applyTransform(input, config);
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("snowPipe is required")) { /* add snowPipe config or make input bounded */ }
  throw e;
}

When it happens

Trigger: Applying the Snowflake write SchemaTransform to an unbounded (streaming, e.g. Kafka-backed) PCollection while the snowPipe configuration field is absent or empty.

Common situations: Switching a pipeline from a bounded batch source to a streaming source without adding the SnowPipe config; running the same config in a streaming runner (Flink/Beam streaming) where isBounded() flips to UNBOUNDED.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• flushRowLimit must be greater than 0.
• flushTimeLimitMillis must be greater than 0.
• shardsNumber must be greater than 0.
• Either table or query must be specified.
• Failed to setLoginTimeout

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/70c2bb4312707ea0. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/SnowflakeWriteSchemaTransformProvider.java:117

              configuration.getDatabase(),
              configuration.getSchema(),
              configuration.getWarehouse(),
              configuration.getRole());

      SnowflakeIO.Write<Row> write =
          SnowflakeIO.<Row>write()
              .withDataSourceConfiguration(dataSourceConfiguration)
              .withStagingBucketName(configuration.getStagingBucketName())
              .withStorageIntegrationName(configuration.getStorageIntegrationName())
              .withUserDataMapper(row -> row.getValues().toArray());

      boolean streaming = rows.isBounded() == PCollection.IsBounded.UNBOUNDED;

      if (streaming) {
        String snowPipe = configuration.getSnowPipe();

        if (snowPipe == null || snowPipe.isEmpty()) {
          throw new IllegalArgumentException("snowPipe is required for streaming writes.");
        }

        write = write.withSnowPipe(snowPipe);

        Integer flushRowLimit = configuration.getFlushRowLimit();
        if (flushRowLimit != null) {
          write = write.withFlushRowLimit(flushRowLimit);
        }

        Long flushTimeLimitMillis = configuration.getFlushTimeLimitMillis();
        if (flushTimeLimitMillis != null) {
          write = write.withFlushTimeLimit(Duration.millis(flushTimeLimitMillis));
        }

        Integer shardsNumber = configuration.getShardsNumber();
        if (shardsNumber != null) {
          write = write.withShardsNumber(shardsNumber);
        }

View on GitHub (pinned to 12126d8942)