Airbyte ETL Hexagonal Architecture — ออกแบบ Data Pipeline ที่ยืดหยุ่น

Airbyte Hexagonal Architecture

Airbyte ETL Hexagonal Architecture Ports Adapters Connector Design Testing Deployment Production Data Pipeline

Connector Design

=== Hexagonal Connector Design ===

from dataclasses import dataclass

Port (Interface)

@abstractmethod

@abstractmethod

@abstractmethod

@abstractmethod

@abstractmethod

@abstractmethod

@abstractmethod

Adapter (Implementation)

self.client = httpx.Client(base_url=base_url,

headers={"Authorization": f"Bearer {auth_token}"})

resp = self.client.get("/health")

return resp.status_code == 200

page = 1

while True:

resp = self.client.get(f"/{stream}", params={"page": page})

เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน CDK Construct Production Setup Guide

data = resp.json()

if not data["items"]: break

yield from data["items"]

page += 1

Core (Business Logic)

class ETLCore:

transformer: ITransformer,

แนะนำเพิ่มเติม — SiamCafeBook

writer: IDestinationWriter):

self.reader = reader

self.transformer = transformer

self.writer = writer

records = []

if self.transformer.validate(record):

transformed = self.transformer.transform(record, schema)

records.append(transformed)

written = self.writer.write_records(records, stream)

return SyncResult(read=len(records), written=written)

@dataclass

class ConnectorComponent:

component: str

layer: str

responsibility: str

example: str

components = [

ConnectorComponent("ISourceReader", "Port",

"Define how to read from any source",

"connect(), read_records(), get_streams()"),

เนื้อหาเกี่ยวข้อง — ทำความเข้าใจ Passkeys WebAuthn Blue Green Canary Deploy

ConnectorComponent("RestApiSourceAdapter", "Adapter",

"Implement REST API reading with pagination",

"HTTP client, auth, pagination logic"),

ConnectorComponent("DatabaseSourceAdapter", "Adapter",

"Implement database reading with cursor",

"SQL query, cursor-based incremental"),

ConnectorComponent("ETLCore", "Core",

"Orchestrate read → transform → write",

"Validation, transformation rules, sync logic"),

ConnectorComponent("IDestinationWriter", "Port",

"Define how to write to any destination",

"write_records(), close(), create_table()"),

ConnectorComponent("BigQueryWriterAdapter", "Adapter",

"Implement BigQuery writing",

แนะนำเพิ่มเติม — สัญญาณเทรดรายวัน XM Signal

"BigQuery client, batch insert, schema mapping"),

]

Testing Strategy

=== Testing Pyramid ===

Unit Test (Core Logic)

transformer = DefaultTransformer()

record = {"name": "test", "value": 42}

schema = {"name": "string", "value": "integer"}

เนื้อหาเกี่ยวข้อง — ทำความเข้าใจ C# Blazor Blue Green Canary Deploy

result = transformer.transform(record, schema)

assert result["name"] == "test"

assert result["value"] == 42

transformer = DefaultTransformer()

record = {"name": None, "value": "not_a_number"}

assert transformer.validate(record) == False

Integration Test (Adapter)

@pytest.fixture

with PostgresContainer("postgres:15") as pg:

yield pg

adapter = DatabaseSourceAdapter(postgres_container.get_connection_url())

adapter.connect({})

records = list(adapter.read_records("users"))

assert len(records) > 0

E2E Test (Full Pipeline)

source = MockSourceAdapter(records=[...])

transformer = DefaultTransformer()

dest = MockDestinationAdapter()

core = ETLCore(source, transformer, dest)

result = core.run_sync("users", schema)

assert result.written == result.read

@dataclass

class TestStrategy:

test_type: str

scope: str

tools: str

เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน additional domain controller คือ

speed: str

when: str

strategies = [

TestStrategy("Unit Test", "Core logic only, mocked ports",

"pytest + Mock/MagicMock", "< 1 sec", "Every commit"),

TestStrategy("Contract Test", "Adapter implements Port correctly",

"pytest + ABC enforcement", "< 5 sec", "Every commit"),

TestStrategy("Integration Test", "Adapter + real external system",

"pytest + Testcontainers", "30-60 sec", "Every PR"),

TestStrategy("Acceptance Test (CAT)", "Full Airbyte connector standard",

"Airbyte CAT framework", "5-10 min", "Before release"),

TestStrategy("E2E Test", "Full pipeline source → destination",

"pytest + Docker Compose", "5-15 min", "Before release"),

]

เคล็ดลับ

• Ports: ออกแบบ Port ให้เป็น Interface ก่อน แล้วค่อย Implement Adapter
• Test: Core Logic ต้องมี Unit Test 100% Coverage ไม่พึ่ง External
• CDK: ใช้ Airbyte CDK สร้าง Custom Connector ตาม Standard
• Incremental: ใช้ Incremental Sync เสมอ ลด Cost และ Duration
• Monitor: ตั้ง Alert สำหรับ Sync Failure Data Quality Cost Spike

Hexagonal Architecture คืออะไร

Ports Adapters แยก Business Logic จาก Infrastructure Interface Adapter Testable Flexible Maintainable Unit Test Mock

สรุป

Airbyte ETL Hexagonal Architecture Ports Adapters Connector CDK Testing Unit Integration E2E Deployment Kubernetes Production