Tối ưu hóa quy trình phân tích dữ liệu với Apache Airflow và Python

python -m venv airflow_env
source airflow_env/bin/activate
pip install apache-airflow

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator

# Khai báo các hằng số cấu hình hệ thống
DEFAULT_ARGS = {
    "owner": "data_team",
    "depends_on_past": False,
    "email_on_failure": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

DAG_ID = "automated_data_analysis_pipeline"
SCHEDULE_INTERVAL = "@daily"


def extract_raw_data(**context):
  """Hàm thực hiện trích xuất dữ liệu thô từ nguồn ngoài."""
  print("Đang tiến hành trích xuất dữ liệu thô...")
  # Logic kết nối API hoặc Database tại đây
  extracted_records = [
      {"id": 1, "value": 100},
      {"id": 2, "value": 200},
      {"id": 3, "value": 300},
  ]
  return extracted_records


def analyze_extracted_data(**context):
  """Hàm thực hiện các tác vụ phân tích số liệu thống kê."""
  ti = context["ti"]
  raw_data = ti.xcom_pull(task_ids="extract_data_task")
  print("Đang thực hiện phân tích dữ liệu...")

  # Tính toán giá trị trung bình đơn giản
  total_value = sum(item["value"] for item in raw_data)
  average_value = total_value / len(raw_data) if raw_data else 0

  analysis_result = {
      "total_records": len(raw_data),
      "average": average_value,
  }
  print(f"Kết quả phân tích: {analysis_result}")
  return analysis_result


def save_analysis_results(**context):
  """Hàm lưu trữ kết quả phân tích vào hệ thống đích."""
  ti = context["ti"]
  analysis_data = ti.xcom_pull(task_ids="analyze_data_task")
  print(f"Đang lưu trữ kết quả lên kho lưu trữ: {analysis_data}")
  # Logic lưu vào Database hoặc Cloud Storage tại đây


# Khởi tạo định nghĩa DAG chính
with DAG(
    dag_id=DAG_ID,
    default_args=DEFAULT_ARGS,
    description="Đường ống tự động hóa phân tích dữ liệu hàng ngày",
    schedule_interval=SCHEDULE_INTERVAL,
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

  # Định nghĩa các tác vụ (Tasks) sử dụng PythonOperator
  extract_task = PythonOperator(
      task_id="extract_data_task",
      python_callable=extract_raw_data,
  )

  analyze_task = PythonOperator(
      task_id="analyze_data_task", python_callable=analyze_extracted_data
  )

  save_task = PythonOperator(
      task_id="save_data_task", python_callable=save_analysis_results
  )

  # Thiết lập luồng thực thi phụ thuộc giữa các tác vụ
  extract_task >> analyze_task >> save_task

from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator


def validate_data_quality(**context):
  """Hàm kiểm tra tính hợp lệ của dữ liệu trước khi phân tích."""
  # Giả lập danh sách bản ghi dữ liệu đầu vào
  sample_records = [{"id": 1, "price": 150.0}, {"id": 2, "price": -10.0}]

  invalid_records = [item for item in sample_records if item["price"] < 0]
  if invalid_records:
    raise ValueError(f"Phát hiện dữ liệu không hợp lệ: {invalid_records}")

  print("Dữ liệu đạt chuẩn chất lượng, sẵn sàng phân tích.")


# Khởi tạo định nghĩa DAG kiểm tra dữ liệu
with DAG(
    dag_id="data_validation_pipeline",
    start_date=datetime(2026, 1, 1),
    schedule_interval="@daily",
    catchup=False,
) as dag:

  validate_task = PythonOperator(
      task_id="validate_data_task", python_callable=validate_data_quality
  )

from datetime import datetime
import pandas as pd
from airflow import DAG
from airflow.operators.python import PythonOperator


def aggregate_sales_data(**context):
  """Hàm sử dụng Pandas để tổng hợp doanh thu theo danh mục sản phẩm."""
  # Giả lập dữ liệu bán hàng thô
  raw_sales_data = {
      "category": ["Electronics", "Clothing", "Electronics", "Clothing"],
      "sales": [500.0, 150.0, 300.0, 200.0],
  }
  df = pd.DataFrame(raw_sales_data)

  # Tổng hợp doanh thu theo từng danh mục
  summary_df = df.groupby("category")["sales"].sum().reset_index()
  print("Kết quả tổng hợp doanh thu theo danh mục:")
  print(summary_df)

  return summary_df.to_dict(orient="records")


# Khởi tạo định nghĩa DAG tổng hợp dữ liệu
with DAG(
    dag_id="sales_aggregation_pipeline",
    start_date=datetime(2026, 1, 1),
    schedule_interval="@weekly",
    catchup=False,
) as dag:

  aggregate_task = PythonOperator(
      task_id="aggregate_sales_task", python_callable=aggregate_sales_data
  )