> ## 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 ingestion patterns

> Learn how to ingest data into RisingWave: continuous streaming from Kafka and CDC, one-time batch loads from S3 and Iceberg, and periodic ingestion.

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.

RisingWave offers several methods for data ingestion, each tailored to different use cases. This guide covers the main patterns to help you choose the best approach for your needs.

For a detailed comparison of core objects like Source, Table, Materialized View, and Sink, see our guide on [Source, Table, MV, and Sink](/get-started/source-table-mv-sink).

## Supported sources

Below is a complete list of source connectors in RisingWave. Click a connector name to see the SQL syntax, options, and a sample statement for connecting RisingWave to the connector.

| Connector                                                          | Version            |
| :----------------------------------------------------------------- | :----------------- |
| [Kafka](/ingestion/sources/kafka)                                  | 3.1.0 or later     |
| [Redpanda](/integrations/sources/redpanda)                         | Latest             |
| [Pulsar](/integrations/sources/pulsar)                             | 2.8.0 or later     |
| [Kinesis](/integrations/sources/kinesis)                           | Latest             |
| [PostgreSQL CDC](/ingestion/sources/postgresql/pg-cdc)             | 10, 11, 12, 13, 14 |
| [MySQL CDC](/ingestion/sources/mysql/mysql-cdc)                    | 5.7, 8.0           |
| [SQL Server CDC](/ingestion/sources/sql-server-cdc)                | 2019, 2022         |
| [MongoDB CDC](/ingestion/sources/mongodb-cdc)                      |                    |
| [CDC via Kafka](/ingestion/cdc-with-risingwave)                    |                    |
| [Google Pub/Sub](/integrations/sources/google-pub-sub)             |                    |
| [Amazon S3](/integrations/sources/s3)                              | Latest             |
| [Google Cloud Storage](/integrations/sources/google-cloud-storage) | Latest             |
| [Azure Blob](/integrations/sources/azure-blob)                     | Latest             |
| [NATS JetStream](/integrations/sources/nats-jetstream)             |                    |
| [MQTT](/integrations/sources/mqtt)                                 |                    |
| [Webhook](/integrations/sources/webhook)                           | Built-in           |
| [Events API (HTTP)](/integrations/sources/events-api)              | External service   |
| [Apache Iceberg](/ingestion/sources/iceberg)                       |                    |
| [Snowflake](/ingestion/sources/snowflake)                          | Latest             |
| [Load generator (datagen)](/ingestion/sources/datagen)             | Built-in           |

For information on supported data formats and encodings, and whether you need to use `CREATE SOURCE` or `CREATE TABLE` with each format, see [Data formats and encoding options](/ingestion/formats-and-encoding-options).

## Continuous streaming ingestion

**What it is**: Real-time, continuous data ingestion from streaming sources that automatically updates as new data arrives.

**When to use**: For real-time analytics, event-driven applications, live dashboards, and when you need immediate data freshness.

### Option: HTTP ingestion (Webhook / Events API)

If you want to ingest events over HTTP without introducing Kafka or another message broker, you can use one of these options:

* **Webhook connector**: RisingWave serves as the webhook destination and ingests requests into `connector = 'webhook'` tables (supports provider-style request validation/signatures). See [Ingest data from webhook](/integrations/sources/webhook).
* **Events API**: Run a standalone service that ingests JSON/NDJSON over HTTP and can execute SQL over HTTP. See [Events API](/integrations/sources/events-api).

### Example: Kafka streaming ingestion

```sql theme={null}
-- Create a streaming table that continuously ingests from Kafka
CREATE TABLE user_events (
  user_id VARCHAR,
  event_type VARCHAR,
  timestamp TIMESTAMP,
  data JSONB
) WITH (
  connector = 'kafka',
  topic = 'user-events',
  properties.bootstrap.server = 'localhost:9092',
  scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

-- Create a materialized view for real-time analytics
CREATE MATERIALIZED VIEW user_activity_summary AS
SELECT
  user_id,
  COUNT(*) as event_count,
  MAX(timestamp) as last_activity
FROM user_events
GROUP BY user_id;
```

### Example: Database CDC (Change Data Capture)

```sql theme={null}
-- Create a shared source for Postgres CDC
CREATE SOURCE shared_source WITH (
    connector='postgres-cdc',
    hostname='localhost',
    port='5432',
    username='postgres',
    password='password',
    database.name='ecommerce',
    schema.name='public' -- Optional, defaults to 'public'
);

-- Create a table from the shared source to continuously ingest changes from the `orders` table in Postgres
CREATE TABLE orders_cdc (
  order_id INT,
  customer_id INT,
  amount DECIMAL,
  status VARCHAR,
  created_at TIMESTAMP
) 
FROM shared_source TABLE 'public.orders';
```

### Example: Message queues (MQTT, NATS, Pulsar)

```sql theme={null}
-- Ingest from MQTT broker
CREATE TABLE iot_data (
  device_id VARCHAR,
  temperature DECIMAL,
  humidity DECIMAL,
  timestamp TIMESTAMP
) WITH (
  connector = 'mqtt',
  url = 'tcp://localhost:1883',
  topic = 'sensors/+/data',
  qos = 'at_least_once'
) FORMAT PLAIN ENCODE JSON;
```

## One-time batch ingestion

**What it is**: Loading data once from external sources like databases, data lakes, or files.

**When to use**: For initial data loads, historical data import, or when you need to load static datasets.

### Example: Batch load from a database

The `postgres-cdc` connector can be used to perform a one-time snapshot of a PostgreSQL table. For other databases, such as MySQL, you can use the corresponding CDC connector and set `snapshot.mode` to `initial_only`.

```sql theme={null}
-- Create a table that takes a snapshot of a PostgreSQL table.
CREATE TABLE historical_sales (
  sale_id INT,
  product_id INT,
  quantity INT,
  sale_date DATE,
  amount DECIMAL
) WITH (
  connector = 'postgres-cdc',
  hostname = 'localhost',
  port = '5432',
  username = 'postgres',
  password = 'password',
  database.name = 'analytics',
  schema.name = 'public',
  table.name = 'sales_history',
  snapshot.mode = 'initial_only' -- Perform a snapshot and then stop.
);
```

### Example: Load from cloud storage (S3, GCS, Azure)

```sql theme={null}
-- Create a source from S3 files
CREATE SOURCE s3_logs (
  log_time timestamptz,
  user_id VARCHAR,
  action VARCHAR
) WITH (
  connector = 's3',
  s3.region_name = 'us-west-2',
  s3.bucket_name = 'my-logs-bucket',
  s3.path = 'logs/2024/'
) FORMAT PLAIN ENCODE CSV;

-- Create a table and load data once
CREATE TABLE processed_logs AS
SELECT * FROM s3_logs;

```

### Example: Load from a data lake (Iceberg)

```sql theme={null}
-- Create a source from an existing Iceberg table
CREATE SOURCE iceberg_data WITH (
  connector = 'iceberg',
  warehouse.path = 's3://my-data-lake/warehouse',
  database.name = 'analytics',
  table.name = 'customer_data',
  catalog.type = 'glue'
);

-- Create a table and load data once
CREATE TABLE customers AS
SELECT * FROM iceberg_data;
```

## Periodic ingestion with external orchestration

**What it is**: RisingWave doesn't have a built-in scheduler, but you can achieve periodic ingestion using external orchestration tools like Cron or Airflow.

**When to use**: For scheduled data updates, daily/hourly batch processing, or when you need precise control over ingestion timing.

### Example: Incremental loading pattern

A common pattern is to use a control table to track the last load time and only ingest new data.

```sql theme={null}
-- A control table to track the last load time for each source
CREATE TABLE load_control (
  table_name VARCHAR PRIMARY KEY,
  last_load_timestamp TIMESTAMP
);

-- Your incremental load script, to be run periodically by an external scheduler
INSERT INTO target_table
SELECT * FROM source_table
WHERE updated_at > (
  SELECT COALESCE(last_load_timestamp, '1970-01-01')
  FROM load_control
  WHERE table_name = 'target_table'
);

-- Update the control table after each successful load
INSERT INTO load_control (table_name, last_load_timestamp)
VALUES ('target_table', NOW())
ON CONFLICT (table_name)
DO UPDATE SET last_load_timestamp = NOW();
```

## Other ingestion methods

### Direct data insertion

You can always insert data directly into a standard table using the `INSERT` statement.

```sql theme={null}
-- Create a standard table
CREATE TABLE transactions (
  id INT,
  amount DECIMAL,
  timestamp TIMESTAMP
);

-- Manually insert data
INSERT INTO transactions VALUES
(1, 100.50, '2024-01-01 10:00:00'),
(2, 250.75, '2024-01-01 10:05:00');
```

### Test data generation

For development and testing, you can use the built-in `datagen` connector to generate mock data streams.

```sql theme={null}
-- Generate a stream of test data
CREATE TABLE test_users (
  id INT,
  name VARCHAR,
  email VARCHAR,
  created_at TIMESTAMP
) WITH (
  connector = 'datagen',
  fields.id.kind = 'sequence',
  fields.id.start = '1',
  fields.name.kind = 'random',
  fields.name.length = '10',
  fields.email.kind = 'random',
  fields.email.length = '15',
  datagen.rows.per.second = '100'
) FORMAT PLAIN ENCODE JSON;
```

## Ingestion method support matrix

| **Data Source**          | **Continuous Streaming** | **One-Time Batch** | **Periodic** | **Notes**                                                                                     |
| ------------------------ | ------------------------ | ------------------ | ------------ | --------------------------------------------------------------------------------------------- |
| **Apache Kafka**         | ✅                        | ❌                  | ⚠️           | Streaming only; periodic via external tools                                                   |
| **Redpanda**             | ✅                        | ❌                  | ⚠️           | Streaming only; periodic via external tools                                                   |
| **Apache Pulsar**        | ✅                        | ❌                  | ⚠️           | Streaming only; periodic via external tools                                                   |
| **AWS Kinesis**          | ✅                        | ❌                  | ⚠️           | Streaming only; periodic via external tools                                                   |
| **Google Pub/Sub**       | ✅                        | ❌                  | ⚠️           | Streaming only; periodic via external tools                                                   |
| **NATS JetStream**       | ✅                        | ❌                  | ⚠️           | Streaming only; periodic via external tools                                                   |
| **MQTT**                 | ✅                        | ❌                  | ⚠️           | Streaming only; periodic via external tools                                                   |
| **PostgreSQL CDC**       | ✅                        | ✅                  | ⚠️           | CDC for streaming; direct connection for batch                                                |
| **MySQL CDC**            | ✅                        | ✅                  | ⚠️           | CDC for streaming; direct connection for batch                                                |
| **SQL Server CDC**       | ✅                        | ✅                  | ⚠️           | CDC for streaming; direct connection for batch                                                |
| **MongoDB CDC**          | ✅                        | ✅                  | ⚠️           | CDC for streaming; direct connection for batch                                                |
| **AWS S3**               | ❌                        | ✅                  | ⚠️           | Batch only; periodic via external tools                                                       |
| **Google Cloud Storage** | ❌                        | ✅                  | ⚠️           | Batch only; periodic via external tools                                                       |
| **Azure Blob**           | ❌                        | ✅                  | ⚠️           | Batch only; periodic via external tools                                                       |
| **Apache Iceberg**       | ❌                        | ✅                  | ⚠️           | Batch only; periodic via external tools                                                       |
| **Snowflake**            | ❌                        | ✅                  | ✅            | Manual one-time load via `REFRESH TABLE`; periodic refresh when `refresh_interval_sec` is set |
| **Datagen**              | ✅                        | ❌                  | ❌            | Test data generation only                                                                     |
| **Direct INSERT**        | ❌                        | ✅                  | ⚠️           | Manual insertion; periodic via external tools                                                 |
| **Webhook**              | ✅                        | ✅                  | ⚠️           | Push-based HTTP ingestion; best for SaaS webhooks + request validation/signatures             |
| **Events API (HTTP)**    | ✅                        | ✅                  | ⚠️           | Run Events API service; supports NDJSON ingestion and SQL over HTTP                           |

**Legend:**

* ✅ **Natively Supported**: Built-in support for this ingestion method.
* ❌ **Not Supported**: This ingestion method is not available for this source.
* ⚠️ **External Tools Required**: Requires external orchestration tools (e.g., Cron, Airflow).

## Best practices

### Choose the right method

* **Streaming**: Use for real-time requirements and continuous data flows.
* **Batch**: Use for historical data, large one-time loads, or static datasets.
* **Periodic**: Use for scheduled updates with external orchestration tools.

### Performance considerations

* Streaming ingestion offers the best real-time performance.
* Batch loading is efficient for large datasets.
* Use materialized views to pre-compute and store results for fast querying.

### Data consistency

* CDC provides high-fidelity replication of database changes.
* For message queues, understand the delivery guarantees (e.g., at-least-once) of your system.
* Use transactions for atomic operations when inserting data manually.
* Monitor data quality and set up alerts.

### Monitoring and operations

* Monitor streaming lag for real-time sources to ensure data freshness.
* Track batch job success and failure rates.
* Set up alerts for data quality issues.
* Use RisingWave's system tables and dashboards for monitoring.

## See also

* [What is a Source?](/reference/what-is-source) — Source concepts, connectors, and source vs. table
* [What is CDC?](/reference/what-is-cdc) — Native CDC connectors for PostgreSQL, MySQL, SQL Server, MongoDB
* [Data processing](/processing/overview) — Transform ingested data with materialized views
* [Data delivery](/delivery/overview) — Send processed results to downstream systems
* [Source, Table, MV, and Sink](/get-started/source-table-mv-sink) — Core object comparison