airbyte-connection-setup

Airbyte Connection Setup Expert

Expert in configuring and managing connections in Airbyte.

Core principles

Connection architecture

• Source-destination pairs: Data streams with transformation
• Incremental synchronization: Preferable to full refresh
• Schema evolution: Handling schema changes
• Failure recovery: Reliable error handling

Synchronization modes

• Full Refresh: Full data overwrite
• Incremental Append: Appending new records
• Incremental Deduped: Deduplication by primary key

Docker Compose Setup

version: '3.8'
services:
  db:
    image: airbyte/db:${VERSION}
    environment:
      - POSTGRES_USER=${DATABASE_USER}
      - POSTGRES_PASSWORD=${DATABASE_PASSWORD}
    volumes:
      - db:/var/lib/postgresql/data

  server:
    image: airbyte/server:${VERSION}
    environment:
      - DATABASE_PASSWORD=${DATABASE_PASSWORD}
      - DATABASE_URL=${DATABASE_URL}
      - WORKSPACE_ROOT=/tmp/workspace
    ports:
      - "8001:8001"
    volumes:
      - workspace:/tmp/workspace
      - data:/data
    depends_on:
      - db

  webapp:
    image: airbyte/webapp:${VERSION}
    ports:
      - "8000:80"
    depends_on:
      - server

Source configuration

# PostgreSQL источник
source_config = {
    "host": "localhost",
    "port": 5432,
    "database": "production_db",
    "username": "airbyte_user",
    "password": "secure_password",
    "ssl_mode": {"mode": "require"},
    "replication_method": {
        "method": "CDC",
        "plugin": "pgoutput",
        "initial_waiting_seconds": 300
    }
}

Destination configuration

# Snowflake назначение
destination_config = {
    "host": "account.snowflakecomputing.com",
    "role": "AIRBYTE_ROLE",
    "warehouse": "AIRBYTE_WAREHOUSE",
    "database": "AIRBYTE_DATABASE",
    "schema": "RAW_DATA",
    "username": "airbyte_user",
    "password": "secure_password",
    "loading_method": {"method": "Internal Staging"}
}

Synchronization configuration

{
  "syncCatalog": {
    "streams": [
      {
        "stream": {
          "name": "users",
          "supportedSyncModes": ["full_refresh", "incremental"]
        },
        "config": {
          "syncMode": "incremental",
          "cursorField": ["updated_at"],
          "destinationSyncMode": "append_dedup",
          "primaryKey": [["id"]]
        }
      }
    ]
  },
  "schedule": {
    "units": 1,
    "timeUnit": "hours"
  }
}

API integration

import requests

# Создание источника
source_payload = {
    "sourceDefinitionId": "decd338e-5647-4c0b-adf4-da0e75f5a750",
    "connectionConfiguration": source_config,
    "workspaceId": workspace_id,
    "name": "Production PostgreSQL"
}

response = requests.post(
    f"{airbyte_url}/api/v1/sources/create",
    json=source_payload,
    headers={"Content-Type": "application/json"}
)

# Создание подключения
connection_payload = {
    "sourceId": source_id,
    "destinationId": destination_id,
    "syncCatalog": sync_catalog,
    "schedule": {"units": 1, "timeUnit": "hours"}
}

requests.post(
    f"{airbyte_url}/api/v1/connections/create",
    json=connection_payload
)

Troubleshooting

Common issues

1. Network connection: Check firewall and access
2. Authentication: Check permissions and expiration
3. Resource limits: Monitor memory and CPU
4. Schema changes: Update configurations

Optimization

• Configuring batch sizes
• Using connection pooling
• Indexing cursor fields
• Performance monitoring

Security

• Storing secrets in vault
• SSL/TLS for all connections
• Network segmentation
• Regular updates