Hacktoberfest 2026: những issue maintainer đã đánh dấu cho tháng Mười, đang mở và phù hợp người mới. Xem issue Hacktoberfest

dags/marketplace_ingestion_dag.py

Đang mở
  • Có gì không ổn?
#17 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Chưa có ai nhận issue này.

Đánh giá

Độ khó
5/5
Thời gian dự kiến
Hơn một tuần
Mức phù hợp với người mới
20/100
Loại issue
Tính năng
Độ rõ ràng
Cần làm rõ
Mức độ hoạt động
Đình trệ
Công nghệ
gcp, pandas, python
Lĩnh vực
cloud, data-engineering

Hướng nghiên cứu

Bắt đầu bằng việc xem xét dags/marketplace_ingestion_dag.py và các module được tham chiếu trong connectors/ và etl_transforms/, đặc biệt là etl_transforms/data_cleaner.py và etl_transforms/bigquery_loader.py. Xác nhận cách Airflow sẽ truy cập các file này và cài đặt các dependency trong môi trường GCP. Công việc được xem là hoàn tất khi workflow trích xuất, biến đổi bằng Pandas và tải vào BigQuery được lập tài liệu có thể chạy dưới dạng DAG hằng ngày.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

การติดตั้ง Airflow (DevOps/CI/CD): Freelancer ต้องมั่นใจว่า Airflow Environment บน GCP (Cloud Composer หรือ VM) ถูกตั้งค่าและสามารถติดตั้ง Dependencies ที่จำเป็นได้
การเชื่อมต่อ Airflow และ Code: ต้องกำหนดวิธีการที่ Airflow จะเข้าถึงโค้ดในโฟลเดอร์ connectors/ และ etl_transforms/ (ส่วนใหญ่มักจะใช้ Docker Image หรือ Volume Sharing)
สร้างโค้ด Transform จริง: สร้างไฟล์ใน etl_transforms/data_cleaner.py เพื่อใช้ Pandas ในการทำความสะอาดและจัดรูปแบบข้อมูลให้เป็นมาตรฐานเดียวกัน ก่อนโหลดเข้า BigQueryfrom airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

--- 1. CONFIGURATION ---

กำหนด Argument พื้นฐานของ DAG

default_args = {
'owner': 'airflow',
'depends_on_past': False,
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
# สำหรับการดึงข้อมูลแบบ Incremental:
'start_date': datetime(2025, 12, 1),
}

--- 2. IMPORT CONNECTOR & TRANSFORM SCRIPTS ---

ต้อง Import ฟังก์ชันหลักจากไฟล์ที่ Freelancer สร้างไว้ใน connectors/ และ etl_transforms/

(ต้องมั่นใจว่า path ใน Airflow สามารถเข้าถึงโค้ดเหล่านี้ได้)

สมมติว่าไฟล์ connector และ transform ถูก Import เข้ามาใน Airflow Environment แล้ว

from connectors.shopee_connector import get_shopee_orders
from connectors.lazada_connector import get_lazada_orders
from etl_transforms.data_cleaner import clean_and_transform
from etl_transforms.bigquery_loader import load_to_bigquery # ฟังก์ชันสำหรับ Task 1.3

--- 3. DEFINE DAG ---

with DAG(
dag_id='marketplace_data_ingestion_v1',
default_args=default_args,
description='ETL pipeline for Shopee, Lazada, and TikTok data ingestion',
schedule_interval=timedelta(hours=24), # รันทุกวัน
catchup=False,
tags=['etl', 'data_ingestion', 'bigquery'],
) as dag:

# === TASK GROUP 1: DATA EXTRACTION (Task 1.1) ===

t1_shopee = PythonOperator(
    task_id='extract_shopee_orders',
    python_callable=get_shopee_orders,
    # ต้องกำหนด op_kwargs เพื่อส่ง parameter (เช่น วันที่เริ่มต้น/สิ้นสุด) ไปให้ฟังก์ชัน
    op_kwargs={'start_date': '{{ ds }}', 'end_date': '{{ next_ds }}'}, 
)

t1_lazada = PythonOperator(
    task_id='extract_lazada_orders',
    python_callable=get_lazada_orders,
    op_kwargs={'start_date': '{{ ds }}', 'end_date': '{{ next_ds }}'},
)

# ... เพิ่ม t1_tiktok และ t1_jsterp ...

# === TASK GROUP 2: DATA TRANSFORMATION (Task 1.2) ===

t2_clean_data = PythonOperator(
    task_id='clean_and_transform_raw_data',
    # ฟังก์ชันนี้จะดึง Raw Data จาก GCS (ที่ Task 1.1 บันทึกไว้) มาประมวลผลด้วย Pandas
    python_callable=clean_and_transform,
    op_kwargs={'execution_date': '{{ ds }}'},
)

# === TASK GROUP 3: DATA LOADING (Task 1.3) ===

t3_load_to_dwh = PythonOperator(
    task_id='load_transformed_data_to_bigquery',
    # ฟังก์ชันนี้จะใช้ Airflow BigQuery Operator หรือโค้ด Python เพื่อโหลดเข้า BigQuery
    python_callable=load_to_bigquery,
    op_kwargs={'target_table': 'marketplace_orders_cleaned'},
)


# --- 4. DEFINE WORKFLOW ORDER ---
# กำหนดลำดับการทำงาน

# 1. ดึงข้อมูลทั้งหมดพร้อมกัน
[t1_shopee, t1_lazada] >> t2_clean_data

# 2. ทำความสะอาดข้อมูลเสร็จแล้วจึงโหลดเข้า DWH
t2_clean_data >> t3_load_to_dwh
Ngôn ngữ chính
HTML
Star
205
Fork
34
Không có pull request nào được merge trong 30 ngày

Chuẩn bị môi trường

Dự án này không cung cấp dev container, Dockerfile hay hướng dẫn đóng góp, nên bạn cần tự thiết lập môi trường: hãy bắt đầu từ README và xem hướng dẫn đóng góp lần đầu của chúng tôi để biết các bước chung.

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của BaseMax/GitHub-Repository-Export-List

Tất cả issue của BaseMax/GitHub-Repository-Export-List

Issue tương tự

Thêm issue về Cloud

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.