Stage and Transform Data with Data Flows

This tutorial demonstrates how to use data flows to build a multi-step extract, load, and transform (ELT) pipeline entirely within the database. You create staging tables, apply transformations across multiple steps, load results into a final table, and clean up temporary resources using a FINALLY clause. Data flows eliminate the need for external orchestration tools by allowing you to execute multiple SQL statements sequentially as a single unit. This approach simplifies pipeline logic, reduces round-trips between the application and database, and ensures cleanup always occurs, even when errors happen. For details about data flow syntax and supported statements, see Data Flows.

Prerequisites

Before you begin, ensure that you have:

A running Ocient deployment with access to execute SQL statements.
The CREATE TABLE and INSERT privileges on the target schema.

Prepare the Target Table

For the purposes of this tutorial, assume a source table named raw_orders with these columns.

Column	Type
`order_id`	INT
`customer_id`	INT
`product_id`	INT
`quantity`	INT
`unit_price`	DOUBLE
`order_date`	TIMESTAMP
`region`	VARCHAR
`status`	VARCHAR

Create the target table that stores the final transformed results. This table holds aggregated order summaries by customer and region.

SQL

CREATE TABLE order_summary (
    customer_id INT,
    region VARCHAR,
    total_orders INT,
    total_revenue DOUBLE,
    avg_order_value DOUBLE,
    last_order_date TIMESTAMP
);

Build and Execute the ELT Pipeline

Build the Data Flow

Create a void data flow that stages raw data, applies transformations, and loads the results into the target table. The data flow performs these operations:

Creates a temporary staging table #valid_orders to hold filtered records.
Creates a temporary aggregation table #customer_agg to hold computed summaries.
Filters out canceled orders and inserts valid records into the staging table.
Computes aggregated metrics per customer and region.
Loads the aggregated results into the final order_summary table.

SQL

BEGIN DATAFLOW
    -- Create temporary tables for staging and aggregation
    CREATE TABLE #valid_orders (
        order_id INT,
        customer_id INT,
        product_id INT,
        quantity INT,
        unit_price DOUBLE,
        order_date TIMESTAMP,
        region VARCHAR
    );

    CREATE TABLE #customer_agg (
        customer_id INT,
        region VARCHAR,
        total_orders INT,
        total_revenue DOUBLE,
        avg_order_value DOUBLE,
        last_order_date TIMESTAMP
    );

    -- Filter out canceled orders into staging
    INSERT INTO #valid_orders
        SELECT order_id, customer_id, product_id, quantity, unit_price, order_date, region
        FROM raw_orders
        WHERE status <> 'canceled';

    -- Compute aggregated metrics per customer and region
    INSERT INTO #customer_agg
        SELECT
            customer_id,
            region,
            COUNT(*) AS total_orders,
            SUM(quantity * unit_price) AS total_revenue,
            SUM(quantity * unit_price) / COUNT(*) AS avg_order_value,
            MAX(order_date) AS last_order_date
        FROM #valid_orders
        GROUP BY customer_id, region;

    -- Load aggregated results into the final table
    INSERT INTO order_summary
        SELECT * FROM #customer_agg;

END DATAFLOW;

Temporary tables use the # prefix and the database scopes them to the data flow block. The system drops them automatically when the data flow completes.

Verify the Results

Query the target table to verify that the data loaded correctly.

SQL

SELECT * FROM order_summary ORDER BY total_revenue DESC LIMIT 10;

Add Error Handling

Optionally, you can wrap the transformation steps in the TRY and EXCEPTION WHEN element to catch and handle errors gracefully. This example logs the error for diagnostics, then re-raises it so the data flow still aborts on failure.

SQL

BEGIN DATAFLOW
    CREATE TABLE #valid_orders (
        order_id INT,
        customer_id INT,
        quantity INT,
        unit_price DOUBLE,
        order_date TIMESTAMP,
        region VARCHAR
    );

    TRY
        INSERT INTO #valid_orders
            SELECT order_id, customer_id, quantity, unit_price, order_date, region
            FROM raw_orders
            WHERE status <> 'canceled';

        INSERT INTO order_summary
            SELECT
                customer_id,
                region,
                COUNT(*),
                SUM(quantity * unit_price),
                SUM(quantity * unit_price) / COUNT(*),
                MAX(order_date)
            FROM #valid_orders
            GROUP BY customer_id, region;
    EXCEPTION WHEN OTHERS THEN
        -- Log the error for diagnostics, then re-raise to abort the data flow
        INSERT INTO error_log VALUES (CURRENT_TIMESTAMP, @SQLERRM, @SQLSTATE);
        RAISE @SQLERRM;
    END TRY;

END DATAFLOW;

Data Flows Build Iterative Queries with Data Flows Validate Data Before Loading with Data Flows Data Definition Language (DDL) Statement Reference Data Manipulation Language (DML) Statement Reference

​Prerequisites

​Prepare the Target Table

​Build and Execute the ELT Pipeline

​Related Links

Prerequisites

Prepare the Target Table

Build and Execute the ELT Pipeline

Related Links