> ## Documentation Index > Fetch the complete documentation index at: https://docs.risingwave.com/llms.txt > Use this file to discover all available pages before exploring further. # Data delivery overview > Deliver streaming results from RisingWave to downstream systems including Kafka, Iceberg, PostgreSQL, Snowflake, ClickHouse, and S3 via sink connectors. Need help generating SQL? Use [Claude Code](https://claude.ai/claude-code) or [Cursor](https://cursor.com) with the [RisingWave MCP server](https://github.com/risingwavelabs/risingwave-mcp) to generate and run SQL interactively. To stream data out of RisingWave, you must create a sink. A sink is an external target that you can send data to. Use the [CREATE SINK](/sql/commands/sql-create-sink) statement to create a sink. You need to specify what data to be exported, the format, and the sink parameters. If you prefer a **pull-based** approach (no sink connector / no external message queue) and want your application to consume changes directly via the Postgres protocol, see [Subscribe to real-time updates](/serve/subscription). Sinks become visible right after you create them, regardless of the backfilling status. Therefore, it's important to understand that the data in the sinks may not immediately reflect the latest state of their upstream sources due to the latency of the sink, connector, and backfilling process. To determine whether the process is complete and the data in the sink is consistent, refer to [Monitor statement progress](/operate/monitor-statement-progress). Currently, RisingWave supports the following sink connectors: | Sink Connector | Connector Parameter | | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------- | | [Apache Doris](/integrations/destinations/apache-doris) | `connector = 'doris'` | | [Apache Iceberg](/integrations/destinations/apache-iceberg) | `connector = 'iceberg'` | | [AWS Kinesis](/integrations/destinations/aws-kinesis) | `connector = 'kinesis'` | | [Cassandra and ScyllaDB](/integrations/destinations/cassandra-or-scylladb) | `connector = 'cassandra'` | | [ClickHouse](/integrations/destinations/clickhouse) | `connector = 'clickhouse'` | | [CockroachDB](/integrations/destinations/cockroachdb) | `connector = 'jdbc'` | | [Delta Lake](/integrations/destinations/delta-lake) | `connector = 'deltalake'` | | [Elasticsearch](/integrations/destinations/elasticsearch) | `connector = 'elasticsearch'` | | [Google BigQuery](/integrations/destinations/bigquery) | `connector = 'bigquery'` | | [Google Pub/Sub](/integrations/destinations/google-pub-sub) | `connector = 'google_pubsub'` | | JDBC: [MySQL](/integrations/destinations/mysql), [PostgreSQL](/integrations/destinations/postgresql), [SQL Server](/integrations/destinations/sql-server), [TiDB](/integrations/destinations/tidb) | `connector = 'jdbc'` | | [Kafka](/integrations/destinations/apache-kafka) | `connector = 'kafka'` | | [MQTT](/integrations/destinations/mqtt) | `connector = 'mqtt'` | | [NATS](/integrations/destinations/nats-and-nats-jetstream) | `connector = 'nats'` | | [Pulsar](/integrations/destinations/apache-pulsar) | `connector = 'pulsar'` | | [Redis](/integrations/destinations/redis) | `connector = 'redis'` | | [Amazon Redshift](/integrations/destinations/redshift) | `connector = 'redshift'` | | [Snowflake](/integrations/destinations/snowflake) | `connector = 'snowflake'` | | [Snowflake v2](/integrations/destinations/snowflake-v2) | `connector = 'snowflake_v2'` | | [StarRocks](/integrations/destinations/starrocks) | `connector = 'starrocks'` | | [Microsoft SQL Server](/integrations/destinations/sql-server) | `connector = 'jdbc'` or `'sqlserver'` | ## Sink decoupling Typically, sinks in RisingWave operate in a blocking manner. This means that if the downstream target system experiences performance fluctuations or becomes unavailable, it can potentially impact the stability of the RisingWave instance. However, sink decoupling can be implemented to address this issue. Sink decoupling introduces a buffering queue between a RisingWave sink and the downstream system. This buffering mechanism helps maintain the stability and performance of the RisingWave instance, even when the downstream system is temporarily slow or unavailable. The `sink_decouple` session variable can be specified to enable or disable sink decoupling. The default value for the session variable is `default`. To enable sink decoupling for all sinks created in the sessions, set `sink_decouple` as `true` or `enable`. ```sql theme={null} SET sink_decouple = true; ``` To disable sink decoupling, set `sink_decouple` as `false` or `disable`, regardless of the default setting. ```sql theme={null} SET sink_decouple = false; ``` Sink decoupling is enabled by default for **all sinks** in RisingWave. When enabled, sinks commit data less frequently to improve stability and reduce load on downstream systems. The commit frequency is controlled by `commit_checkpoint_interval`: * **Decoupled sinks (default)**: Commit every 10 checkpoints (approximately 10 seconds) * **Non-decoupled sinks**: Commit every checkpoint (approximately 1 second) * **Iceberg sinks**: Commit every 60 checkpoints (approximately 60 seconds) by default This means that with sink decoupling enabled, data may take up to 10-60 seconds to become visible in the downstream system, even though RisingWave's internal checkpoints occur every second. Iceberg's exactly-once semantics require sink decoupling to be enabled. If you disable sink decoupling for an Iceberg sink, exactly-once will be automatically disabled. An internal system table `rw_sink_decouple` is provided to query whether a created sink has enabled sink decoupling or not. ```sql theme={null} dev=> select sink_id, is_decouple from rw_sink_decouple; sink_id | is_decouple ---------+------------- 2 | f 5 | t (2 rows) ``` ## Upsert sinks and primary keys For each sink, you can specify the data format. The available data formats are `upsert`, `append-only`, and `debezium`. To determine which data format is supported by each sink connector, please refer to the detailed guide listed above. In the `upsert` sink, a non-null value updates the last value for the same key or inserts a new value if the key doesn't exist. A NULL value indicates the deletion of the corresponding key. When creating an `upsert` sink, note whether or not you need to specify the primary key in the following situations. * If the downstream system supports primary keys and the table in the downstream system has a primary key, you must specify the primary key with the `primary_key` field when creating an upsert JDBC sink. * If the downstream system supports primary keys but the table in the downstream system has no primary key, then RisingWave does not allow users to create an upsert sink. A primary key must be defined in the table in the downstream system. * If the downstream system does not support primary keys, then users must define the primary key when creating an upsert sink. ## Delivery semantics Different sink connectors provide different delivery guarantees: ### Exactly-once semantics * **Iceberg**: Supports exactly-once delivery when `is_exactly_once = true` (the default). Requires sink decoupling to be enabled. ### At-least-once semantics The following sinks guarantee at-least-once delivery. Events may be redelivered in case of failures: * **AWS Kinesis**: Due to PutRecords API limitations, provides at-least-once delivery and eventual consistency * **Elasticsearch**: At-least-once with possible redeliveries during failures * **NATS**: At-least-once with possible redeliveries during failures * **Kafka**: Non-transactional writes with at-least-once guarantees via retries * **Most other sinks**: Unless otherwise specified, sinks provide at-least-once semantics For append-only sinks, at-least-once semantics mean duplicate records may appear. For upsert sinks, duplicates are naturally deduplicated by the primary key in the downstream system. ## Sink buffering behavior A sink will buffer incoming data within each barrier interval if the stream's internal primary key (stream key) differs from the user-defined sink primary key and the sink is not `append-only`. This mismatch can cause update events for the same sink key to be split across multiple upstream fragments in a distributed execution, leading to out-of-order operations if sent directly. Buffering allows the sink to compact and reorder updates so that delete events are emitted before insert events, ensuring correct semantics. If the stream key matches the sink primary key exactly, or if the sink is configured as `append-only` or `force_append_only`, no buffering is performed and the sink emits changes immediately as they arrive. ### Force compaction You can set the `force_compaction` option to `true` to enable buffering and compaction even when the stream key matches the sink primary key exactly. This allows you to: * Reduce the number of output messages by compacting updates within each barrier interval * Emit at most one update per key within a barrier interval, simplifying downstream logic Example: ```sql theme={null} CREATE SINK sink_t_force_compaction FROM t_force_compaction WITH ( connector = 'kafka', properties.bootstrap.server = 'message_queue:29092', topic = 'test-rw-force-compaction-sink', force_compaction = 'true', primary_key = 'id' ) FORMAT UPSERT ENCODE JSON; ``` ## Sink data in parquet or json format RisingWave supports sinking data in Parquet or JSON formats to cloud storage services, including [S3](/integrations/destinations/aws-s3), [Google Cloud Storage (GCS)](/integrations/destinations/google-cloud-storage), [Azure Blob Storage](/integrations/destinations/azure-blob), and [WebHDFS](/integrations/destinations/webhdfs). This eliminates the need for complex data lake setups. Once the data is saved, the files can be queried using RisingWave's batch processing engine through the `file_scan` API. You can also leverage third-party OLAP query engines to enhance data processing capabilities. Below is an example to sink data to S3: ```sql theme={null} CREATE SINK test_file_sink FROM test WITH ( connector = 's3', s3.region_name = '{config['S3_REGION']}', s3.bucket_name = '{config['S3_BUCKET']}', s3.credentials.access = '{config['S3_ACCESS_KEY']}', s3.credentials.secret = '{config['S3_SECRET_KEY']}', s3.endpoint_url = 'https://{config['S3_ENDPOINT']}' s3.path = '', type = 'append-only', force_append_only='true' ) FORMAT PLAIN ENCODE PARQUET(force_append_only='true'); ``` File sink currently supports only append-only mode, so please change the query to `append-only` and specify this explicitly after the `FORMAT ... ENCODE ...` statement. ## Batching strategy for file sink Added in v2.1.0. RisingWave implements batching strategies for file sinks, including [S3](/integrations/destinations/aws-s3), [Google Cloud Storage (GCS)](/integrations/destinations/google-cloud-storage), [Azure Blob Storage](/integrations/destinations/azure-blob), and [WebHDFS](/integrations/destinations/webhdfs). This optimizes file management by preventing the generation of numerous small files. The batching strategy is available for Parquet, JSON, and CSV encode. ### Defaults By default, RisingWave finalizes the current file and starts a new one when **either** of the following thresholds is reached: | Parameter | Default | Description | | ------------------ | ------- | --------------------------------------------------------- | | `rollover_seconds` | `10` | Finalize the file after this many seconds of writing. | | `max_row_count` | `10240` | Finalize the file after this many rows have been written. | Because either threshold can trigger a file rollover, a large initial snapshot or backfill will typically be spread across **multiple files**. For example, a table with 16,000 rows will produce at least two Parquet files (≈ 10,240 rows in the first file, ≈ 5,760 rows in the second). Inspecting only the first file and seeing roughly 10K rows does **not** mean data is missing — check all objects under the configured sink path and sum their row counts. The condition for batching is relatively coarse-grained. The actual number of rows or exact timing of file completion may vary from the specified thresholds, as this function is intentionally flexible to prioritize efficient file management. ### Category * **Batching based on row numbers**: RisingWave monitors the number of rows written and completes the file once the maximum row count threshold is reached. Specify `max_row_count` option in the `WITH` clause to configure this behavior. * **Batching based on rollover interval**: RisingWave checks the threshold each time a chunk is about to be written and when a barrier is encountered. Specify `rollover_seconds` option in the `WITH` clause to configure this behavior. * A file is finalized when **either** threshold is hit first. ### File organization You can specify `path_partition_prefix` option in the `WITH` clause to organize files into subdirectories based on their creation time. The available options are month, day, or hour. If not specified, files will be stored directly in the root directory without any time-based subdirectories. Files follow the naming pattern `/Option/_.`. `Timestamp` differentiates files batched by the rollover interval. Starting from v2.7.0, the file naming scheme replaces the previous `executor_id` component with a UUID, meaning that further operations should no longer rely on `executor_id` in the filename. The output files look like below: ``` path/2024-09-20/01935d2c-8f12-7890-abcd-ef1234567890_1727072046.parquet path/2024-09-20/01935d2c-8f12-7890-abcd-ef1234567890_1727072055.parquet ``` ### Example ```sql theme={null} CREATE SINK s1 FROM t WITH ( connector = 's3', max_row_count = '100', rollover_seconds = '10', type = 'append-only', path_partition_prefix = 'day' ) FORMAT PLAIN ENCODE PARQUET (force_append_only=true); ``` In this example, a file is finalized as soon as either 100 rows have been written **or** 10 seconds have elapsed — whichever comes first. Once finalized, the file becomes visible in the downstream sink system. ### Verifying row counts after a backfill When a sink is first created, RisingWave performs an initial backfill of the upstream table. Because the default `max_row_count` is `10240`, a table with more than 10,240 rows will produce multiple output files. To verify that all rows were delivered: 1. List all objects under the configured sink path (e.g., with `aws s3 ls s3://// --recursive`). 2. Sum the row counts across all files. 3. Compare the total against the row count of the upstream table. If you need all rows in a single file (for small datasets) or want larger files, increase `max_row_count` and/or `rollover_seconds` when creating the sink: ```sql theme={null} CREATE SINK s1 FROM t WITH ( connector = 's3', s3.region_name = '', s3.bucket_name = '', s3.credentials.access = '', s3.credentials.secret = '', s3.path = 'output/', max_row_count = '1000000', rollover_seconds = '3600', type = 'append-only' ) FORMAT PLAIN ENCODE PARQUET (force_append_only=true); ``` ## See also * [What is a Sink?](/reference/what-is-sink) — Sink concepts, emission modes, and connector overview * [Data ingestion](/ingestion/overview) — How data enters RisingWave * [Data processing](/processing/overview) — Transform data before delivery * [CREATE SINK](/sql/commands/sql-create-sink) — SQL reference