Airbyte

# Airbyte 커넥터 설정 예제 (connection.yaml)
# PostgreSQL에서 BigQuery로 데이터 동기화

version: "0.1.0"

# 소스 커넥터 설정 (PostgreSQL)
source:
  sourceType: postgres
  name: production-postgres
  config:
    host: db.example.com
    port: 5432
    database: analytics
    username: ${POSTGRES_USER}
    password: ${POSTGRES_PASSWORD}
    schemas:
      - public
      - sales
    replication_method:
      method: CDC
      replication_slot: airbyte_slot
      publication: airbyte_publication

# 목적지 커넥터 설정 (BigQuery)
destination:
  destinationType: bigquery
  name: analytics-warehouse
  config:
    project_id: my-gcp-project
    dataset_id: raw_data
    dataset_location: asia-northeast3
    credentials_json: ${BIGQUERY_CREDENTIALS}
    loading_method:
      method: GCS Staging
      gcs_bucket_name: airbyte-staging
      gcs_bucket_path: /staging

# 동기화 설정
connection:
  name: postgres-to-bigquery-sync
  namespaceDefinition: source
  namespaceFormat: "${SOURCE_NAMESPACE}"
  prefix: "raw_"

  # 동기화 스케줄
  schedule:
    scheduleType: cron
    cronExpression: "0 */6 * * *"  # 6시간마다

  # 스트림 설정 (테이블별 동기화 방식)
  syncCatalog:
    streams:
      - stream:
          name: orders
          namespace: sales
        config:
          syncMode: incremental
          cursorField:
            - updated_at
          destinationSyncMode: append_dedup
          primaryKey:
            - - order_id

      - stream:
          name: customers
          namespace: public
        config:
          syncMode: incremental
          cursorField:
            - modified_date
          destinationSyncMode: append_dedup
          primaryKey:
            - - customer_id

      - stream:
          name: products
          namespace: public
        config:
          syncMode: full_refresh
          destinationSyncMode: overwrite

# Airbyte Terraform 설정 예제
# terraform/airbyte.tf
---
resource "airbyte_source_postgres" "production" {
  name         = "production-postgres"
  workspace_id = var.airbyte_workspace_id

  configuration = {
    host     = var.postgres_host
    port     = 5432
    database = "analytics"
    username = var.postgres_user
    password = var.postgres_password

    replication_method = {
      replication_method_cdc = {
        replication_slot = "airbyte_slot"
        publication      = "airbyte_publication"
      }
    }
  }
}

resource "airbyte_destination_bigquery" "warehouse" {
  name         = "analytics-warehouse"
  workspace_id = var.airbyte_workspace_id

  configuration = {
    project_id       = var.gcp_project_id
    dataset_id       = "raw_data"
    dataset_location = "asia-northeast3"
    credentials_json = var.bigquery_credentials

    loading_method = {
      gcs_staging = {
        gcs_bucket_name = "airbyte-staging"
        gcs_bucket_path = "/staging"
      }
    }
  }
}