fivetran-dagster-to-orchestra

Fivetran: Dagster -> Orchestra Conversion

Overview

In Dagster, Fivetran is integrated via dagster-fivetran: a FivetranResource (or FivetranWorkspace) holds the API key/secret and build_fivetran_assets turns each connector into assets. Materializing the assets triggers the connector sync. In Orchestra the equivalent is a single Sync task under the FIVETRAN integration — trigger and wait in one task.

Parameter Mapping

Dagster concept	Orchestra YAML field	Notes
FivetranResource(api_key=, api_secret=)	connection:	Orchestra Fivetran connection (stores API key + secret)
build_fivetran_assets(connector_id=...)	parameters.connector_id	Connector slug — copy verbatim
asset materialization	(always waits)	Orchestra always waits for completion
destination_tables=[...]	(not needed)	Whole connector synced
asset key / name	name:	Human-readable task name
upstream asset deps	depends_on:	Upstream task names/UUIDs

Orchestra YAML Structure

version: v1
name: <pipeline-name>
pipeline:
  <stage-uuid>:
    tasks:
      <task-uuid>:
        integration: FIVETRAN
        integration_job: FIVETRAN_SYNC_ALL
        name: <descriptive name>
        connection: <orchestra-fivetran-connection-name>
        parameters:
          connector_id: <fivetran-connector-id>
        depends_on: []
        condition: null
        tags: []

Conversion Steps

1. Find the Fivetran resource — locate FivetranResource and the connector_id passed to build_fivetran_assets.
2. Verify/create the Orchestra connection — Settings -> Connections -> Fivetran with API key + secret. Note its name.
3. Replace assets with a task block — one task per connector.
4. Wire dependencies — convert upstream asset deps to depends_on:.

Before / After Example

Dagster (before)

from dagster import Definitions, EnvVar
from dagster_fivetran import FivetranResource, build_fivetran_assets

fivetran = FivetranResource(
    api_key=EnvVar("FIVETRAN_API_KEY"), api_secret=EnvVar("FIVETRAN_API_SECRET"),
)
salesforce_assets = build_fivetran_assets(
    connector_id="bronzing_regularly",
    destination_tables=["accounts", "opportunities"],
)
defs = Definitions(assets=salesforce_assets, resources={"fivetran": fivetran})