Pulumi IaC กับ Data Pipeline ETL — วิธีใช้

Pulumi สำหรับ Data Pipeline

Pulumi ใช้ภาษาโปรแกรมจริง Python TypeScript สร้าง Cloud Infrastructure ETL Pipeline ประกอบด้วย Extract Transform Load ดึงข้อมูลแปลงโหลดเข้า Data Warehouse

เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: GCP Pub Sub GitOps Workflow

ใช้ Pulumi สร้าง S3 Buckets, Glue Jobs, Redshift Cluster, Airflow Environment อัตโนมัติ Version Control ด้วย Git Reproducible ทุกครั้ง

เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: Spdr Stock — คู่มือฉบับสมบูรณ์ 2026

Airflow Orchestration

# === Airflow DAG สำหรับ ETL Pipeline ===

# dags/etl_daily_sales.py
# from airflow import DAG
# from airflow.operators.python import PythonOperator
# from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
# from airflow.providers.amazon.aws.operators.s3 import S3CopyObjectOperator
# from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
# from datetime import datetime, timedelta
#
# default_args = {
#     'owner': 'data-team',
#     'depends_on_past': False,
#     'email_on_failure': True,
#     'email': ['data-alerts@company.com'],
#     'retries': 2,
#     'retry_delay': timedelta(minutes=5),
# }
#
# with DAG(
#     'etl_daily_sales',
#     default_args=default_args,
#     description='Daily Sales ETL Pipeline',
#     schedule_interval='0 6 * * *',  # ทุกวัน 6:00 AM
#     start_date=datetime(2024, 1, 1),
#     catchup=False,
#     tags=['etl', 'sales', 'daily'],
# ) as dag:
#
#     wait_for_data = S3KeySensor(
#         task_id='wait_for_raw_data',
#         bucket_name='etl-pipeline-raw',
#         bucket_key='orders/{{ ds }}/*.parquet',
#         timeout=3600,
#     )
#
#     extract_api = PythonOperator(
#         task_id='extract_from_api',
#         python_callable=extract_orders,
#         op_kwargs={'date': '{{ ds }}'},
#     )
#
#     transform = GlueJobOperator(
#         task_id='transform_data',
#         job_name='etl-pipeline-transform',
#         script_args={
#             '--date': '{{ ds }}',
#             '--raw_bucket': 'etl-pipeline-raw',
#             '--processed_bucket': 'etl-pipeline-processed',
#         },
#     )
#
#     load_redshift = PythonOperator(
#         task_id='load_to_redshift',
#         python_callable=load_to_redshift,
#         op_kwargs={'date': '{{ ds }}'},
#     )
#
#     notify = PythonOperator(
#         task_id='send_notification',
#         python_callable=send_slack_notification,
#     )
#
#     wait_for_data >> extract_api >> transform >> load_redshift >> notify

# Pulumi — สร้าง MWAA (Managed Airflow) Environment
# mwaa = aws.mwaa.Environment(f"{project}-airflow",
#     name=f"{project}-airflow-{env}",
#     airflow_version="2.8.1",
#     execution_role_arn=airflow_role.arn,
#     source_bucket_arn=dags_bucket.arn,
#     dag_s3_path="dags/",
#     environment_class="mw1.small",
#     max_workers=5,
#     network_configuration=aws.mwaa.EnvironmentNetworkConfigurationArgs(
#         security_group_ids=[sg.id],
#         subnet_ids=private_subnets,
#     ),
# )

dag_structure = {
    "DAG": "etl_daily_sales",
    "Schedule": "0 6 * * * (Daily 6AM)",
    "Tasks": [
        "wait_for_raw_data (S3 Sensor)",
        "extract_from_api (Python)",
        "transform_data (Glue Job)",
        "load_to_redshift (Python)",
        "send_notification (Slack)",
    ],
}

print("Airflow DAG:")
print(f"  Name: {dag_structure['DAG']}")
print(f"  Schedule: {dag_structure['Schedule']}")
print(f"  Tasks:")
for task in dag_structure['Tasks']:
    print(f"    - {task}")

Best Practices

Pulumi Stacks: แยก Stack ต่อ Environment (dev/staging/prod)
Secrets: ใช้ Pulumi Config Secrets เก็บ Password และ API Keys
Idempotent: ทำ ETL Jobs ให้ Idempotent รันซ้ำได้ผลเหมือนเดิม
Partitioning: แบ่ง Data เป็น Partitions ตาม Date ลด Scan Time
Monitoring: ตั้ง Alerts สำหรับ Failed Jobs, Data Quality Issues
Testing: เขียน Unit Tests สำหรับ Transform Functions ทดสอบก่อน Deploy

Pulumi คืออะไร

IaC Tool ใช้ภาษาโปรแกรมจริง Python TypeScript Go แทน DSL สร้าง Cloud Resources AWS Azure GCP State Management Preview Changes Drift Detection

แนะนำเพิ่มเติม — อ่านเพิ่มเติมที่ SiamCafeBook

เนื้อหาเกี่ยวข้อง — อ่านต่อ: Crowdsec IPS Technical Debt Management — คู่มือฉบับสมบูรณ์ 2026