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 or Cursor with the RisingWave MCP server 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.

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	3.1.0 or later
Redpanda	Latest
Pulsar	2.8.0 or later
Kinesis	Latest
PostgreSQL CDC	10, 11, 12, 13, 14
MySQL CDC	5.7, 8.0
SQL Server CDC	2019, 2022
MongoDB CDC
CDC via Kafka
Google Pub/Sub
Amazon S3	Latest
Google Cloud Storage	Latest
Azure Blob	Latest
NATS JetStream
MQTT
Webhook	Built-in
Events API (HTTP)	External service
Apache Iceberg
Snowflake	Latest
Load generator (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.

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.
• Events API: Run a standalone service that ingests JSON/NDJSON over HTTP and can execute SQL over HTTP. See Events API.

Example: Kafka streaming ingestion

-- 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)

-- 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)

-- 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

For PostgreSQL and MySQL, use the postgres_query or mysql_query table-valued function to query the upstream database on demand. Unlike a CDC connector, each invocation executes the upstream query once and does not continue reading database changes.

-- Run the PostgreSQL query once and store its result in RisingWave.
CREATE TABLE historical_sales AS
SELECT *
FROM postgres_query(
  'localhost',
  '5432',
  'postgres',
  '<password>',
  'analytics',
  'SELECT sale_id, product_id, quantity, sale_date, amount FROM public.sales_history;'
);

CDC connectors are intended for continuous replication. Do not use a Debezium snapshot mode as a substitute for a finite batch load.

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

When a batch query such as CREATE TABLE AS or INSERT ... SELECT reads an object-storage source, RisingWave lists matching objects once for that query. In streaming queries, the same source continues discovering new files instead.

-- 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.credentials.access = '<your-access-key>',
  s3.credentials.secret = '<your-secret-key>',
  match_pattern = 'logs/2024/*.csv'
) 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)

-- 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;

The batch query loads the current Iceberg snapshot once. To continuously ingest new snapshots from an append-only Iceberg table, use the source in a streaming query instead.

Periodic ingestion

What it is: Reloading or incrementally importing external data on a schedule.

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

S3, GCS, Iceberg, and Snowflake tables can use refresh_mode = 'FULL_RELOAD'. Set refresh_interval_sec for native periodic refresh, or omit it and run REFRESH TABLE when needed. Full-reload refresh is currently in technical preview.

Use an external orchestrator such as Cron or Airflow for connectors without native refresh support or for custom incremental-loading logic.

Example: Externally orchestrated incremental load

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

-- 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.

-- 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.

-- 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 for streaming; postgres_query for one-time batch loads
MySQL	✅	✅	⚠️	CDC for streaming; mysql_query for one-time batch loads
SQL Server CDC	✅	❌	⚠️	CDC for streaming; no native one-time batch query
MongoDB CDC	✅	❌	⚠️	CDC for streaming; no native one-time batch query
AWS S3	✅	✅	✅	Append-only file discovery; batch queries; optional full reload
Google Cloud Storage	✅	✅	✅	Append-only file discovery; batch queries; optional full reload
Azure Blob	✅	✅	⚠️	Append-only file discovery and batch queries; external orchestration for scheduled reloads
Apache Iceberg	✅	✅	✅	Append-only continuous ingestion; batch queries; optional full reload
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 native full-reload scheduling where supported, or external orchestration for custom incremental loads and other connectors.

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? — Source concepts, connectors, and source vs. table
• What is CDC? — Native CDC connectors for PostgreSQL, MySQL, SQL Server, MongoDB
• Data processing — Transform ingested data with materialized views
• Data delivery — Send processed results to downstream systems
• Source, Table, MV, and Sink — Core object comparison