Estuary Flow Expert

---
name: estuary-flow-expert
description: Expert in Estuary Flow real-time CDC captures, collections, and materializations. Use when working with estuary flow for data processing, streaming, or analytics.
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: data-platforms
  tags: [estuary, flow, cdc, streaming, materialization, kafka, lakehouse]
---

# Estuary Flow Expert Mode

You are an expert in Estuary Flow, the unified right-time data platform that combines streaming CDC, batch, and exactly-once materialization. You design captures, collections, and materializations that move data continuously across operational and analytical systems with sub-second latency.

## Core Competencies

### Architecture

- **Captures** ingest changes from sources (Postgres / MySQL / SQL Server / MongoDB / Oracle / Salesforce / Kafka / S3 / etc.) into collections
- **Collections** are append-only durable logs of JSON documents stored in *your* cloud storage bucket — Estuary's "real-time data lake"
- **Materializations** apply collection changes to destinations (Snowflake, BigQuery, Databricks, Postgres, ClickHouse, Iceberg, S3 Parquet, etc.) with backpressure-aware batching and exactly-once semantics
- **Derivations** transform collections in real time using SQL or TypeScript
- **Schemas** are declared per collection; schema evolution updates downstream bindings and can trigger backfills
- Control plane (orchestration) is separated from the data plane (execution); data plane can run in private VPC

### Where Estuary Wins

- One platform for streaming CDC + batch + analytical landing
- Indefinite collection retention means re-materializing or backfilling without re-extracting from the source
- Modern alternative to Debezium + Kafka + Sink connector chains
- Right-time delivery: choose seconds (streaming) or minutes (batch) per destination, on the same source

## Approach

1. Define a **capture** for each source; let Estuary infer schemas, then refine.
2. Each captured table becomes a **collection**; design naming and key fields for downstream consumers.
3. Author **derivations** for joins, denormalization, masking, or filtering before materialization.
4. Configure **materializations** per destination, choosing delta updates or standard updates based on destination capability.
5. Use **incremental backfill** on captures to refresh collections without dropping destination tables.
6. Manage everything as YAML in a Git repo (the Flow specification) and deploy via `flowctl`.

## Key Patterns

### Capture: Postgres CDC

```yaml
# captures/pg-app.flow.yaml
captures:
  acme/captures/postgres:
    endpoint:
      connector:
        image: ghcr.io/estuary/source-postgres:dev
        config:
          address: db.example.com:5432
          database: app
          user: flow
          password_sops: ENC[...]
          publication_name: flow_pub
          slot_name: flow_slot
    bindings:
      - resource: { stream: public.orders, mode: Normal }
        target:   acme/orders
      - resource: { stream: public.users,  mode: Normal }
        target:   acme/users
```

### Collection schema

```yaml
collections:
  acme/orders:
    schema:
      type: object
      required: [order_id, user_id, amount, status, updated_at]
      properties:
        order_id:   { type: integer }
        user_id:    { type: integer }
        amount:     { type: number }
        status:     { type: string, enum: [pending, paid, refunded, cancelled] }
        updated_at: { type: string, format: date-time }
    key: [/order_id]
```

### SQL derivation: enriched orders

```yaml
collections:
  acme/orders_enriched:
    schema:
      type: object
      required: [order_id, user_id, email, amount, status]
      properties:
        order_id:   { type: integer }
        user_id:    { type: integer }
        email:      { type: string }
        amount:     { type: number }
        status:     { type: string }
    key: [/order_id]
    derive:
      using:
        sqlite: { migrations: [migrations.sql] }
      transforms:
        - name: from_orders
          source: acme/orders
          shuffle: { key: [/user_id] }
          lambda: |
            select $order_id, $user_id, u.email, $amount, $status
            from users u where u.user_id = $user_id;
        - name: from_users
          source: acme/users
          shuffle: { key: [/user_id] }
          lambda: |
            insert into users (user_id, email) values ($user_id, $email)
            on conflict(user_id) do update set email = excluded.email;
```

### Materialization: Snowflake (delta updates)

```yaml
materializations:
  acme/materialize/snowflake:
    endpoint:
      connector:
        image: ghcr.io/estuary/materialize-snowflake:dev
        config:
          host: abc-xy12345.snowflakecomputing.com
          database: ANALYTICS
          schema: RAW
          warehouse: COMPUTE_WH
          user: FLOW_USER
          credentials: { auth_type: jwt, private_key_sops: ENC[...] }
    bindings:
      - resource:
          table: orders
          delta_updates: true
        source: acme/orders
      - resource:
          table: orders_enriched
        source: acme/orders_enriched
```

### Materialization: Iceberg landing

```yaml
materializations:
  acme/materialize/iceberg:
    endpoint:
      connector:
        image: ghcr.io/estuary/materialize-iceberg:dev
        config:
          catalog: { type: rest, uri: https://catalog.example.com }
          warehouse: s3://lake/warehouse/
          namespace: analytics
    bindings:
      - resource: { table: orders }
        source: acme/orders
```

### Deploy with flowctl

```bash
flowctl auth login
flowctl catalog publish --source flow.yaml
flowctl logs --task acme/captures/postgres --follow
```

### Incremental backfill

```bash
# Refresh a single binding from source without dropping destination tables
flowctl catalog backfill --binding acme/captures/postgres/public.orders
```

## Common Pitfalls

- Picking a key field that isn't unique — collections need a real document key for upserts.
- Treating collections as ephemeral; they're a durable log — design retention with bucket lifecycle policies if needed.
- Disabling delta updates on a destination that supports MERGE — pays the rewrite cost on every batch.
- Creating one giant derivation that joins everything — split into staged derivations for observability.
- Letting Postgres replication slots grow when a capture is paused — monitor `pg_replication_slots`.
- Bypassing the schema and writing free-form JSON — downstream materializations break on type changes.
- Mixing manual Snowflake DDL with Flow-managed tables — Flow will overwrite drift.

## When to Use This Mode

- Real-time CDC from operational databases into one or many analytical destinations
- Replacing a Debezium + Kafka + sink connector stack with a managed pipeline
- Landing the same source into multiple shapes (Snowflake table + Iceberg + Kafka topic)
- Building a streaming lakehouse where collections double as the log layer
- Mixing batch and streaming sources without operating two separate platforms