airbyte-server-airflow-to-orchestra

Airbyte Server: Airflow → Orchestra Conversion

Overview

Self-hosted Airbyte uses the same AirbyteTriggerSyncOperator / AirbyteSensor as Airbyte Cloud in Airflow, but the Airflow connection points to your own host. In Orchestra the equivalent is a Sync task under the AIRBYTE_SERVER integration, which uses a separate connection type that stores the server host URL.

Parameter Mapping

Airflow parameter	Orchestra YAML field	Notes
airbyte_conn_id	connection:	The name of the Orchestra connection to Airbyte Server (stores host + API key)
connection_id	parameters.connection_id	Airbyte connection UUID — copy verbatim
asynchronous / AirbyteSensor	(not needed)	Orchestra always polls; drop the sensor
task_id	name:	Human-readable task name
upstream >> chains	depends_on:	List upstream task names/UUIDs

Orchestra YAML Structure

version: v1
name: <pipeline-name>
pipeline:
  <stage-uuid>:
    tasks:
      <task-uuid>:
        integration: AIRBYTE_SERVER
        integration_job: AIRBYTE_SERVER_JOB
        name: <task_id value from Airflow>
        connection: <orchestra-airbyte-server-connection-name>
        parameters:
          connection_id: <airbyte-connection-uuid>
          job_type: sync          # required: "sync" or "reset"
        depends_on: []
        condition: null
        tags: []

Conversion Steps

1. Identify the Airflow task — locate AirbyteTriggerSyncOperator. Note the airbyte_conn_id (which holds the host) and the connection_id UUID.
2. Create/verify the Orchestra connection — in Orchestra Settings → Connections, create a connection of type Airbyte Server pointing to your host URL and API credentials. Note its name.
3. Replace operator with task block — use the YAML above.
4. Drop any paired AirbyteSensor — redundant in Orchestra.
5. Wire dependencies.

Before / After Example

Airflow DAG (before)

from airflow.providers.airbyte.operators.airbyte import AirbyteTriggerSyncOperator
from airflow.providers.airbyte.sensors.airbyte import AirbyteSensor

trigger = AirbyteTriggerSyncOperator(
    task_id="trigger_crm_sync",
    airbyte_conn_id="airbyte_self_hosted",
    connection_id="bb112233-4455-6677-8899-aabbccddeeff",
    asynchronous=True,
)

wait = AirbyteSensor(
    task_id="wait_crm_sync",
    airbyte_conn_id="airbyte_self_hosted",
    airbyte_job_id=trigger.output,
    poke_interval=60,
)

trigger >> wait