Starrocks Etl And Tasks

StarRocks ETL and Tasks

Design ETL pipelines in StarRocks and run them asynchronously or on a schedule with SUBMIT TASK.

When to Use

• Designing batch, incremental, or real-time ETL pipelines
• Running long INSERT/CTAS in the background (avoid client timeout)
• Scheduling recurring transformations natively in StarRocks
• Backfilling or migrating data
• Implementing SCD Type 2 dimensions

Related skills: [starrocks-data-loading] for the underlying load methods; [starrocks-materialized-views] as an alternative to scheduled aggregation tasks; [starrocks-table-design] for staging/target schemas.

ETL Pipeline Patterns

Pattern 1: Daily Batch Load from S3

-- Step 1: Load raw data to staging table
LOAD LABEL mydb.load_staging_{{ ds }}
(
    DATA INFILE("s3://bucket/data/{{ ds }}/*.parquet")
    INTO TABLE staging_events
    FORMAT AS "parquet"
)
WITH BROKER
(
    "aws.s3.access_key" = "KEY",
    "aws.s3.secret_key" = "SECRET",
    "aws.s3.region" = "us-east-1"
)
PROPERTIES (
    "timeout" = "7200"
);

-- Step 2: Wait for load to complete (check status)

-- Step 3: Transform and load to final table
INSERT INTO events PARTITION (p{{ ds_nodash }})
SELECT
    event_id,
    user_id,
    event_type,
    DATE(event_time) as event_date,
    event_time,
    CAST(properties['amount'] AS DECIMAL(18,2)) as amount
FROM staging_events
WHERE DATE(event_time) = '{{ ds }}';

-- Step 4: Clean up staging
TRUNCATE TABLE staging_events;

Pattern 2: Incremental Load with Deduplication

-- Primary key table for automatic deduplication
CREATE TABLE user_profiles (
    user_id BIGINT NOT NULL,
    username VARCHAR(100),
    email VARCHAR(200),
    status VARCHAR(20),
    updated_at DATETIME
)
PRIMARY KEY (user_id)
DISTRIBUTED BY HASH(user_id) BUCKETS 16
PROPERTIES (
    "enable_persistent_index" = "true"
);

-- Stream load with updates (automatic upsert)
curl --location-trusted -u user:password \
    -H "label:user_profile_update_$(date +%s)" \
    -H "format:json" \
    -T incremental_users.json \
    http://fe_host:8030/api/mydb/user_profiles/_stream_load

-- Only latest record per user_id is kept

Pattern 3: Real-time Aggregation Pipeline

-- Step 1: Raw events table with hourly partitioning
CREATE TABLE raw_events (
    event_time DATETIME,
    event_type STRING,
    user_id BIGINT,
    processing_time INT
)
DUPLICATE KEY (event_time)
PARTITION BY date_trunc('hour', event_time)  -- Hourly for high frequency
DISTRIBUTED BY HASH(user_id) BUCKETS 64;

-- Step 2: Continuous loading via Routine Load from Kafka
CREATE ROUTINE LOAD mydb.routine_load_events ON raw_events
COLUMNS(event_time, event_type, user_id, processing_time)
PROPERTIES (
    "desired_concurrent_number" = "3",
    "max_batch_interval" = "20",
    "format" = "json"
)
FROM KAFKA (...);

-- Step 3: Async MV for 5-minute aggregation
CREATE MATERIALIZED VIEW mv_events_5min
REFRESH ASYNC START('2024-01-01 00:00:00') EVERY(INTERVAL 5 MINUTE)
AS
SELECT
    date_trunc('minute', event_time) as time_bucket,  -- Minute-level bucketing
    event_type,
    COUNT(*) as event_count,
    COUNT(DISTINCT user_id) as unique_users,
    AVG(processing_time) as avg_processing_time
FROM raw_events
WHERE event_time >= DATE_SUB(NOW(), INTERVAL 1 DAY)
GROUP BY date_trunc('minute', event_time), event_type;

-- Step 4: Query the MV for dashboard
SELECT * FROM mv_events_5min
WHERE time_bucket >= DATE_SUB(NOW(), INTERVAL 1 HOUR)
ORDER BY time_bucket DESC;

Pattern 4: SCD Type 2 (Slowly Changing Dimension)

UPDATE only works on Primary Key tables, so the history table must be a PK table (with valid_from in the key to keep multiple versions per user_id).

Transaction caveats: Explicit transactions require v3.5+; UPDATE inside a transaction needs a shared-data cluster on v4.0+ and must precede the INSERT on the same table. On other deployments, run the two statements separately instead of wrapping them in BEGIN/COMMIT.

-- Historical dimension table (Primary Key table so UPDATE is supported)
CREATE TABLE dim_users_history (
    user_id BIGINT NOT NULL,
    valid_from DATETIME NOT NULL,
    username VARCHAR(100),
    email VARCHAR(200),
    status VARCHAR(20),
    valid_to DATETIME,
    is_current BOOLEAN
)
PRIMARY KEY (user_id, valid_from)
DISTRIBUTED BY HASH(user_id) BUCKETS 16;

-- Step 1: Close existing current records for updated users
UPDATE dim_users_history
SET valid_to = NOW(), is_current = false
WHERE user_id IN (SELECT user_id FROM staging_users)
  AND is_current = true;

-- Step 2: Insert new current records
INSERT INTO dim_users_history
SELECT
    user_id,
    NOW() as valid_from,
    username,
    email,
    status,
    NULL as valid_to,
    true as is_current
FROM staging_users;

SUBMIT TASK (Async Execution)

SUBMIT TASK runs ETL statements asynchronously in the background. Available since v2.5, with scheduled execution since v3.3.

When to Use SUBMIT TASK

Use SUBMIT TASK when:

• The operation would exceed your session's query_timeout
• You want StarRocks-native scheduling without external orchestrators
• You need to fire-and-forget and check status later
• You're running multiple independent ETL steps in parallel

Don't use SUBMIT TASK when:

• You need the result immediately in the same session
• The operation finishes in seconds (overhead not worth it)
• You already have Airflow/dbt orchestrating the pipeline (use those instead)

Syntax

SUBMIT TASK <task_name>
[SCHEDULE [START(<schedule_start>)] EVERY(INTERVAL <schedule_interval>)]
[PROPERTIES("key" = "value" [, ...])]
AS <etl_statement>

Supported ETL statements:

• CREATE TABLE AS SELECT (v3.0+)
• INSERT INTO / INSERT OVERWRITE (v3.0+)
• CACHE SELECT (v3.3+)

One-off Background Task

-- Background INSERT that won't block your session
SUBMIT TASK backfill_2024
AS
INSERT INTO target_table
SELECT * FROM source_table
WHERE dt >= '2024-01-01' AND dt < '2025-01-01';

Scheduled Recurring Task

-- Refresh aggregation table every 5 minutes
SUBMIT TASK refresh_5min_agg
SCHEDULE EVERY(INTERVAL 5 MINUTE)
AS
INSERT OVERWRITE hourly_metrics
SELECT
    date_trunc('hour', event_time) AS hour,
    event_type,
    COUNT(*) AS cnt,
    SUM(amount) AS total
FROM raw_events
WHERE event_time >= DATE_SUB(NOW(), INTERVAL 2 HOUR)
GROUP BY date_trunc('hour', event_time), event_type;

Scheduled Task with Start Time

-- Start at midnight, run daily
SUBMIT TASK daily_summary
SCHEDULE START('2024-01-01 00:00:00') EVERY(INTERVAL 1 DAY)
AS
INSERT OVERWRITE daily_revenue PARTITION (dt)
SELECT
    DATE(order_time) AS dt,
    store_id,
    SUM(amount) AS revenue
FROM orders
WHERE order_time >= DATE_SUB(CURDATE(), INTERVAL 1 DAY)
  AND order_time < CURDATE()
GROUP BY DATE(order_time), store_id;

Task with Session Properties

-- Extend timeout and enable profiling for a heavy task
SUBMIT TASK heavy_etl
PROPERTIES(
    "session.insert_timeout" = "36000",
    "session.enable_profile" = "true"
)
AS
INSERT INTO warehouse.fact_orders
SELECT * FROM staging.raw_orders;

CTAS as Background Task

-- Create a snapshot table without blocking
SUBMIT TASK create_snapshot
AS
CREATE TABLE snapshot_20240101
AS SELECT * FROM production_table
WHERE dt = '2024-01-01';

Monitoring Tasks

-- List all registered tasks
SELECT task_name, `schedule`, create_time, definition
FROM information_schema.tasks;

-- Check execution history for a specific task
SELECT task_name, state, create_time, finish_time, error_message
FROM information_schema.task_runs
WHERE task_name = 'daily_summary'
ORDER BY create_time DESC
LIMIT 10;

-- Find failed task runs
SELECT task_name, state, error_message, create_time
FROM information_schema.task_runs
WHERE state = 'FAILED'
ORDER BY create_time DESC;

Drop a Task

-- Stop and remove a scheduled task (wrap the name in backticks to avoid parse failures)
DROP TASK `daily_summary`;

FE Configuration Tuning