FlowBridge walkthrough

This walkthrough picks up the compiled DAGDocument produced at the end of the MapCraft walkthrough — the Customer orders ingest project, lane Warehouse. By the end of this page you will have:

Attached a field-level policy that masks customer_email on a specific delivery.
Configured Extract & Load against a Postgres source with incremental CDC.
Configured a Delivery Adapter (SFTP JSON drop).
Run a dry-run and inspected the preview SQL.
Triggered a live run and watched the status transitions.
Set up a daily schedule.
Observed a drift event and accepted its fix proposal.
Built a serverless package for AWS Lambda.

You'll need

The Customer orders ingest project with a compiled DAG (the end-state of the MapCraft walkthrough).
A Postgres connection reference registered in your DataChord workspace (any test database is fine — no production access required for a dry-run).
Either a real SFTP target or a local file path for delivery; the walkthrough shows both.

Step 1 — Attach a policy to `customer_email`

Open /mapcraft/{projectId}/policies. The Policy-as-Code page lists every target field on the left and the destinations on the right. Click Add policy next to customer_email.

Fill in the form:

Field:        customer_email
Action:       Mask
Pattern:      ***@{domain}
Destination:  marketing_export   (created in Step 3 below)

The resulting Policy JSON

policy
{
  "id": "pol_email_mask_marketing",
  "field": "customer_email",
  "action": "mask",
  "pattern": "***@{domain}",
  "destinations": ["marketing_export"]
}

Policies fail closed

If you save a policy that references an unknown destination, the next compile will fail with policy.unknown_destination. That is intentional — DataChord will not silently emit unmasked data. Create the destination first (Step 3), then attach the policy.

Step 2 — Configure Extract & Load

Open /mapcraft/{projectId}/extract-load. You will see a form-style configurator and an Inspect raw config link that shows the same data as JSON.

UI form
Raw JSON

Source reference:       postgres://prod-replica/datachord/customer_orders_raw
Target table:           warehouse.orders_clean
CDC mode:               ◉ incremental    ○ full-refresh
High-water column:      updated_at
Initial high-water:     2026-03-14T00:00:00Z
Batch size:             10,000 rows

extract_load_config
{
  "source_ref": "postgres://prod-replica/datachord/customer_orders_raw",
  "target_table": "warehouse.orders_clean",
  "cdc_mode": "incremental",
  "high_water_column": "updated_at",
  "initial_high_water": "2026-03-14T00:00:00Z",
  "batch_size": 10000
}

Click Save config. The DAG now has an extract_load node wired into its extract stage. The ExtractLoadState row that tracks the watermark is created lazily on the first run.

Backfill vs. incremental

The incremental mode reads rows whose updated_at is greater than the stored high-water. The full-refresh mode replays everything; use it once for the initial load, or trigger it later via the Backfill button on this page if you need to re-derive the target.

Step 3 — Configure a Delivery Adapter

Open /mapcraft/{projectId}/deliver. Click New adapter. The form gives you four adapter types; the walkthrough shows three.

SFTP JSON drop
HTTP webhook
FILE (local, for testing)

Name:           marketing_export
Type:           SFTP
Host:           sftp.partner.example.com
Port:           22
Path template:  /inbound/datachord/orders/{yyyymmdd}.json
Format:         JSON (one record per line)
Secret ref:     vault://datachord/sftp/marketing_export

Never hard-code secrets

The Secret ref field is a pointer into your workspace's secrets backend. DataChord will not accept a raw password or private key in this form. If you have not registered the secret yet, do that first; the adapter will save in disabled status until the reference resolves.

Name:            partner_webhook
Type:            HTTP
URL:             https://partner.example.com/api/orders/ingest
Method:          POST
Headers:         {"Authorization": "Bearer ${vault://datachord/partner/token}"}
Retry policy:    exponential, max 5, base 2s
Format:          JSON (single batched array)

Name:            local_smoke
Type:            FILE
Path template:   /tmp/datachord/orders/{yyyymmdd}.json
Format:          JSON

The FILE adapter is intended for development. Do not use it for production — it writes to the FlowBridge worker's local disk, which is ephemeral.

Click Save. Then click Test adapter — FlowBridge sends a single fake row through the format + adapter pipeline and reports success or a specific failure (e.g. auth-failed, path-not-writable).

Step 4 — Run a dry-run

Open /mapcraft/{projectId} (the DAG view). The toolbar now shows two buttons: Dry-run and Run now.

Click Dry-run. FlowBridge plans the run without writing to the target or delivering anywhere. The DAG view annotates each node with a row count estimate and a previewable SQL plan.

Preview SQL for the transform node

preview · transform
SELECT
  src.order_id                                                AS order_id,
  CASE
    WHEN pol.action = 'mask'
      THEN regexp_replace(src.customer_email, '^[^@]+', '***')
    ELSE src.customer_email
  END                                                         AS customer_email,
  COALESCE(
    to_date(NULLIF(src.order_date, ''), 'YYYY-MM-DD'),
    to_date(NULLIF(src.order_date, ''), 'MM/DD/YYYY'),
    to_date(NULLIF(src.order_date, ''), 'DD-Mon-YYYY')
  )                                                           AS order_date,
  lk.country_code                                             AS country_code,
  src.total_amount                                            AS total_amount,
  src.currency                                                AS currency,
  src.status                                                  AS status
FROM        customer_orders_raw   src
LEFT JOIN   country_iso2_lookup  lk  ON lk.country_name = src.country
LEFT JOIN   policy_bindings      pol ON pol.field = 'customer_email'
                                    AND pol.destination = :destination
WHERE       src.total_amount >= 0
  AND       src.updated_at  >  :high_water;

Notice that the customer_email masking is conditional on the destination. For the warehouse target (warehouse.orders_clean), the policy binding joins to nothing, so the email passes through unmasked. For marketing_export, the binding resolves to mask, so the email is masked.

Step 5 — Trigger a live run

Click Run now. The DAG transitions immediately to pending and the run row appears in the Run history panel.

For the sample data (15 rows), expect the transition pending → running → success within a couple of seconds. The run row shows:

Run ID	Trigger	Status	Records	Duration
`run_abc123`	manual	✅ success	14	1.4s

Note records_processed: 14, not 15 — ORD-1008 was filtered out by the target's minimum: 0 constraint on total_amount. The run did not fail; it correctly excluded the bad row. If you want to be alerted on excluded rows, add a validator (out of scope for this walkthrough).

Step 6 — Inspect run history

The run history panel is paginated. Each row links to a detail view with:

Per-node duration and row counts.
The exact SQL that ran (with bound parameters).
The delivery log (per adapter).
Any error_message and a copy-to-clipboard stack trace if the run failed.

Sample rows:

Run ID	Trigger	Status	Records	Started	Duration
`run_abc123`	manual	✅ success	14	2026-05-16 01:18:02Z	1.4s
`run_abc090`	scheduled	✅ success	8	2026-05-15 02:00:01Z	1.1s
`run_aba991`	drift-fix	✅ success	0	2026-05-14 12:42:31Z	0.6s
`run_aba803`	scheduled	❌ failed	0	2026-05-12 02:00:01Z	0.2s

Step 7 — Set up a daily schedule

Open /mapcraft/{projectId}/schedule. Click Add schedule.

Daily
Hourly
Every 15 min

Cadence:      Cron
Expression:   0 2 * * *      (every day at 02:00 UTC)
Timezone:     UTC
Enabled:      ✔

Cadence:      Cron
Expression:   0 * * * *      (top of every hour, UTC)
Timezone:     UTC
Enabled:      ✔

Cadence:      Interval
Every:        15 minutes
Timezone:     UTC
Enabled:      ✔

Click Save. The schedule is stored as a DAGSchedule and picked up by the FlowBridge beat scheduler within one cadence interval.

Schedules and CDC compose

Because Extract & Load is in incremental mode, each scheduled run reads only the rows added or updated since the previous high-water. You do not need to overlap the cron and the watermark interval — FlowBridge keeps the high-water consistent across runs.

Step 8 — Drift detection

Open /mapcraft/{projectId}/drift. FlowBridge watches the source schema and the freshness of the source table on every run.

Suppose the partner adds a new column to customer_orders_raw. The next run records a DriftEvent and auto-generates a proposal.

Example DriftEvent JSON

drift event
{
  "id": "drift_evt_42",
  "kind": "schema",
  "detected_at": "2026-05-16T02:00:04Z",
  "source": "customer_orders_raw",
  "diff": [
    { "op": "add", "path": "shipping_address", "type": "STRING" }
  ],
  "severity": "info",
  "proposal_id": "drift_prop_42"
}

The proposal page (/drift/proposals/{id}) shows what would change and a single Accept fix button. Accepting it:

Updates the source schema in the Project.
Adds an UNMAPPED row for shipping_address in the Table view.
Triggers a drift-fix run (trigger type drift-fix) to refresh state.

You decide whether to map the new field. The pipeline keeps running either way.

Step 9 — Build a serverless package

Open /mapcraft/{projectId}/package. Click Build.

AWS Lambda
GCP Cloud Run
GCP Cloud Function

Target:        AWS Lambda
Runtime:       python3.12
Memory:        512 MB
Timeout:       60s
Trigger:       EventBridge schedule (mirrored from FlowBridge)

The build produces a zip with the transform module, lookups, policies, and a handler.py entrypoint. The artifact carries a SHA-256 so you can verify it against the registry it lands in.

Target:        Cloud Run
Image base:    gcr.io/datachord/base-python:3.12
Concurrency:   1
Trigger:       Cloud Scheduler (mirrored)

The build produces an OCI image manifest plus the transform module mounted as the entrypoint.

Target:        Cloud Function (2nd gen)
Runtime:       python312
Trigger:       HTTP / Pub/Sub

The build produces a function source bundle ready for gcloud functions deploy.

After the build, the artifact appears in the Artifacts table:

Artifact ID	Target	SHA-256	Status
`pkg_lambda_001`	aws-lambda	`0fa3…b9e2`	built

Click Publish to mark it as registered with your runtime (the registry reference is recorded on the artifact). Status moves from built to published. The artifact is now your reproducible deliverable.

Closing notes

You now have a single MapCraft Project that:

Loads warehouse.orders_clean incrementally from Postgres every day at 02:00 UTC.
Drops a daily SFTP JSON file to a partner with customer_email masked.
Watches the source for drift and proposes fixes.
Has a published AWS Lambda package as a portable artifact.

Pick the lanes you need

You did not have to choose between warehouse load, partner drop, and serverless deliverable — they all came from the same compiled DAG. If tomorrow you need a fourth lane (a webhook to a different partner), that is a new Delivery Adapter, not a new project.

Where to go next

Re-read the Product flow & architecture page now that the objects (SourceProfile, Project, DAGDocument, DAGRun, PackageArtifact) have concrete meaning.
Explore the other FlowBridge surfaces you did not use here: Sync-to-app (OAuth + upsert into SaaS apps) is on the same page as Deliver, under the /sync route.
Bring your real data through StructScan and repeat — the walkthrough scales to projects with hundreds of fields without changing shape.

Step 1 — Attach a policy to customer_email​

Step 2 — Configure Extract & Load​

Step 3 — Configure a Delivery Adapter​

Step 4 — Run a dry-run​

Step 5 — Trigger a live run​

Step 6 — Inspect run history​

Step 7 — Set up a daily schedule​

Step 8 — Drift detection​

Step 9 — Build a serverless package​

Closing notes​

Where to go next​