Skip to main content

Your First ETL Pipeline

Build a complete Extract-Transform-Load pipeline using ProAI.

Architecture​

ext PostgreSQL (Source) → Filter & Aggregate (Transform) → Snowflake (Destination)

Step 1: Create the Pipeline​

ash proai pipeline create user-analytics --template etl

Step 2: Define the Source​

Edit pipelines/user-analytics.yml:

sources:
  - name: raw_users
    type: postgres
    connector: my-postgres
    config:
      table: users
      incremental: true
      cursor_column: updated_at
`

## Step 3: Add Transformations

```yaml
transforms:
  - name: clean_users
    type: sql
    query: |
      SELECT
        id,
        TRIM(email) as email,
        LOWER(status) as status,
        created_at,
        updated_at
      FROM raw_users
      WHERE email IS NOT NULL

  - name: aggregate_metrics
    type: sql
    query: |
      SELECT
        DATE_TRUNC('day', created_at) as date,
        status,
        COUNT(*) as user_count,
        COUNT(DISTINCT email) as unique_emails
      FROM clean_users
      GROUP BY 1, 2
`

## Step 4: Configure the Destination

```yaml
destinations:
  - name: analytics_warehouse
    type: snowflake
    connector: my-snowflake
    config:
      database: ANALYTICS
      schema: PUBLIC
      table: user_metrics
      load_strategy: merge
      merge_keys: [date, status]
`

## Step 5: Run the Pipeline

`ash
proai run user-analytics
`

Monitor the execution in the [Dashboard](https://app.proai.io).

:::tip[Pro Tip]
Use proai run user-analytics --dry-run to validate the pipeline without actually executing it.
:::

## See Also

- [ETL Jobs Overview](/docs/etl-jobs/overview) — Complete ETL documentation
- [Sources](/docs/etl-jobs/sources) — All source types
- [Transformations](/docs/etl-jobs/transformations) — Transform reference