5 ทางเลือก Airflow ชั้นนำสำหรับการจัดระเบียบงานข้อมูล (มีโค้ดตัวอย่าง)

อ่านบล็อก Airflow vs Prefect: เลือกอะไรให้เหมาะกับเวิร์กโฟลว์ข้อมูล เพื่อดูการเปรียบเทียบอย่างละเอียดระหว่างเครื่องมือทั้งสอง

เริ่มต้นใช้งาน Prefect

เริ่มโปรเจกต์ Prefect ด้วยการติดตั้งแพ็กเกจ Python รันคำสั่งต่อไปนี้ในเทอร์มินัล

$ pip install -U prefect

จากนั้นสร้างสคริปต์ Python ชื่อ prefect_etl.py และเขียนโค้ดดังต่อไปนี้

from prefect import task, flow
import pandas as pd

# Extract data
@task
def extract_data():
    # Simulating data extraction
    data = {
        "name": ["Alice", "Bob", "Charlie"],
        "age": [25, 30, 35],
        "city": ["New York", "Los Angeles", "Chicago"]
    }
    df = pd.DataFrame(data)
    return df

# Transform data
@task
def transform_data(df: pd.DataFrame):
    # Example transformation: adding a new column
    df["age_plus_ten"] = df["age"] + 10
    return df

# Load data
@task
def load_data(df: pd.DataFrame):
    # Simulating data load
    print("Loading data to target destination:")
    print(df)

# Defining the flow
@flow(log_prints=True)
def etl():
    raw_data = extract_data()
    transformed_data = transform_data(raw_data)
    load_data(transformed_data)

# Running the flow
if __name__ == "__main__":
    etl()

โค้ดข้างต้นนิยามฟังก์ชันงาน extract_data(), transform_data(), และ load_data() และเรียกทำงานตามลำดับในฟังก์ชันโฟลว์ชื่อ etl() ฟังก์ชันเหล่านี้ถูกสร้างด้วยตัวตกแต่ง (decorators) ของ Prefect

โดยย่อ เราสร้าง pandas DataFrame ปรับแปลง แล้วพิมพ์ผลลัพธ์สุดท้ายด้วย print เป็นการจำลอง ETL pipeline แบบง่าย

ในการรันเวิร์กโฟลว์ ให้รันสคริปต์ Python ด้วยคำสั่งต่อไปนี้

$ python prefect_etl.py 

จะเห็นว่าเวิร์กโฟลว์ของเรารันสำเร็จ

บันทึกการรันโฟลว์ของ Prefect

การดีพลอยโฟลว์

ต่อไปจะดีพลอยเวิร์กโฟลว์ เพื่อให้รันตามตารางเวลาหรือทริกเกอร์จากเหตุการณ์ได้ การดีพลอยยังช่วยให้ติดตามและจัดการหลายเวิร์กโฟลว์แบบศูนย์กลาง

ในการดีพลอยโฟลว์ จะใช้ Prefect CLI โดยคำสั่ง deploy ต้องระบุชื่อไฟล์ Python ชื่อฟังก์ชันโฟลว์ในไฟล์ และชื่อดีพลอยเมนต์ ในที่นี้ตั้งชื่อว่า “simple_etl”

$ prefect deploy prefect_etl.py:etl -n 'simple_etl'

หลังรันคำสั่ง อาจมีข้อความแจ้งว่ายังไม่มี worker pool สำหรับรันดีพลอยเมนต์ ให้สร้าง worker pool ด้วยคำสั่งต่อไปนี้

$ prefect worker start --pool 'datacamp'

เมื่อมี worker pool แล้ว ให้เปิดหน้าต่างเทอร์มินัลอีกอันเพื่อรันดีพลอยเมนต์ คำสั่ง prefect deployment run ต้องการอาร์กิวเมนต์รูปแบบ “<flow-function-name>/<deployment-name>” ดังคำสั่งด้านล่าง

$ prefect deployment run 'etl/simple_etl

เมื่อรันดีพลอยเมนต์ จะมีข้อความแจ้งว่าเวิร์กโฟลว์กำลังรันอยู่ โดย flow run ที่สร้างขึ้นจะถูกสุ่มตั้งชื่อ ในกรณีนี้คือ witty-lorikeet.

Creating flow run for deployment 'etl/simple_etl'...
Created flow run 'witty-lorikeet'.
└── UUID: 4e0495b0-9c7e-4ed8-b9ab-5160994dc7f0
└── Parameters: {}
└── Job Variables: {}
└── Scheduled start time: 2024-06-22 14:05:01 PKT (now)
└── URL: <no dashboard available>

เพื่อดูล็อกทั้งหมด ให้สลับไปที่หน้าต่างเทอร์มินัลที่สตาร์ต worker pool ไว้

สรุปการรันโฟลว์ของ Prefect

ควรสตาร์ตเว็บเซิร์ฟเวอร์ของ Prefect เพื่อแสดงผลการรันโฟลว์ให้เข้าใจง่ายขึ้นและจัดการเวิร์กโฟลว์อื่นๆ

$ prefect server start 

หลังรันคำสั่ง จะถูกพาไปยังแดชบอร์ดของ Prefect โดยอัตโนมัติ หรือเข้าไปที่ http://127.0.0.1:4200 ผ่านเบราว์เซอร์ก็ได้

หน้าจอเว็บเซิร์ฟเวอร์ของ Prefect

แดชบอร์ดช่วยให้รันเวิร์กโฟลว์ใหม่ ดูล็อก ตรวจสอบ work pool ตั้งค่าการแจ้งเตือน และเลือกตัวเลือกขั้นสูงอื่นๆ ได้ ถือเป็นโซลูชันครบถ้วนสำหรับการจัดระเบียบงานข้อมูลยุคใหม่

หากต้องการเรียนรู้การสร้างและรันท่อทาง ML ด้วย Prefect ดูบทช่วยสอน Using Prefect for Machine Learning Workflows ได้

2. Dagster

Dagster เป็นเฟรมเวิร์กโอเพนซอร์สสำหรับวิศวกรข้อมูลในการนิยาม กำหนดตาราง และติดตามท่อทางข้อมูล รองรับการปรับขนาดสูง และเอื้อต่อการทำงานร่วมกันระหว่างทีมข้อมูลหลายทีม 

Dagster ช่วยให้นิยาม data assets เป็นฟังก์ชัน Python ด้วย decorators เมื่อกำหนด asset แล้วสามารถรันได้อย่างราบรื่นผ่านตารางเวลาหรือทริกเกอร์แบบอิงเหตุการณ์

เมื่อเทียบกับ Airflow, Dagster ช่วยให้พัฒนา ทดสอบ และรีวิว pipeline ภายในเครื่องได้ เน้นแนวทาง orchestration แบบอิงสินทรัพย์ (asset-based) และเป็นคลาวด์/คอนเทนเนอร์เนทีฟ

เริ่มต้นใช้งาน Dagster

เราจะสร้าง ETL pipeline แบบง่าย รัน และแสดงผลผ่านเว็บเซิร์ฟเวอร์ของ Dagster คล้ายแดชบอร์ดของ Prefect เว็บเซิร์ฟเวอร์ของ Dagster ช่วยติดตามเวิร์กโฟลว์หลายตัวและตั้งเวลา/จัดการ assets แบบศูนย์กลาง

เริ่มจากติดตั้งแพ็กเกจ Python

$ pip install dagster -q

จากนั้นสร้างฟังก์ชัน Python 3 ฟังก์ชันสำหรับ extract, transform และ load ข้อมูล ฟังก์ชันชื่อ create_dirty_data(),  clean_data(), และ load_cleaned_data() ตามโค้ด โดยใช้ตัวตกแต่ง @asset เพื่อประกาศฟังก์ชันเป็น data assets ใน Dagster

ถัดไปสร้างงานของ asset (ตัวแปร job) ด้วยทุก asset (ตัวแปร all_assets) และสร้าง asset definition (ตัวแปร defs) 

สามารถข้ามส่วน asset definition ได้ แต่จะสำคัญหากต้องการตั้งเวลาการรัน รันหลายงาน และตั้งค่าเซนเซอร์

import pandas as pd
import numpy as np
from dagster import asset, Definitions, define_asset_job, materialize

@asset
def create_dirty_data():
    # Create a sample DataFrame with dirty data
    data = {
        'Name': [' John Doe ', 'Jane Smith', 'Bob Johnson ', '  Alice Brown'],
        'Age': [30, np.nan, 40, 35],
        'City': ['New York', 'los angeles', 'CHICAGO', 'Houston'],
        'Salary': ['50,000', '60000', '75,000', 'invalid']
    }
    df = pd.DataFrame(data)
    
    # Save the DataFrame to a CSV file
    dirty_file_path = 'dag_data/dirty_data.csv'
    df.to_csv(dirty_file_path, index=False)
    
    return dirty_file_path

@asset
def clean_data(create_dirty_data):
    # Read the dirty CSV file
    df = pd.read_csv(create_dirty_data)
    
    # Clean the data
    df['Name'] = df['Name'].str.strip()
    df['Age'] = pd.to_numeric(df['Age'], errors='coerce').fillna(df['Age'].mean())
    df['City'] = df['City'].str.upper()
    df['Salary'] = df['Salary'].replace('[\$,]', '', regex=True)
    df['Salary'] = pd.to_numeric(df['Salary'], errors='coerce').fillna(0)
    
    # Calculate average salary
    avg_salary = df['Salary'].mean()
    
    # Save the cleaned DataFrame to a new CSV file
    cleaned_file_path = 'dag_data/cleaned_data.csv'
    df.to_csv(cleaned_file_path, index=False)
    
    return {
        'cleaned_file_path': cleaned_file_path,
        'avg_salary': avg_salary
    }

@asset
def load_cleaned_data(clean_data):
    cleaned_file_path = clean_data['cleaned_file_path']
    avg_salary = clean_data['avg_salary']
    
    # Read the cleaned CSV file to verify
    df = pd.read_csv(cleaned_file_path)
    
    print({
        'num_rows': len(df),
        'num_columns': len(df.columns),
        'avg_salary': avg_salary
    })

# Define all assets
all_assets = [create_dirty_data, clean_data, load_cleaned_data]

# Create a job that will materialize all assets
job = define_asset_job("all_assets_job", selection=all_assets)

# Create Definitions object
defs = Definitions(
    assets=all_assets,
    jobs=[job]
)

if __name__ == "__main__":
    result = materialize(all_assets)
    print("Pipeline execution result:", result.success)

รันโค้ดข้างบนใน Jupyter Notebook หรือสร้างไฟล์ Python แล้วรันก็ได้ 

เมื่อรันโค้ด จะได้ล็อกการทำงานของเวิร์กโฟลว์ครบถ้วน 

เว็บเซิร์ฟเวอร์ Dagster

เพื่อแสดงผล assets และการรันงาน ต้องติดตั้งและรันเว็บเซิร์ฟเวอร์ของ Dagster เว็บเซิร์ฟเวอร์ช่วยให้รันงาน ทำ materialize ทีละ asset และติดตามหลายงานพร้อมกัน

$ pip install dagster-webserver

ในการเริ่มเซิร์ฟเวอร์ Dagster ให้ใช้ Daster CLI และระบุตำแหน่งไฟล์ Python ในที่นี้ตั้งชื่อไฟล์ว่า dagster_pipe.py.

$ dagster dev -f dagster_pipe.py  

คำสั่งนี้จะเปิดเว็บเซิร์ฟเวอร์ในเบราว์เซอร์โดยอัตโนมัติ หรือจะเข้าโดยตรงที่ http://127.0.0.1:3000 ผ่านเบราว์เซอร์ก็ได้

หน้าจอเว็บเซิร์ฟเวอร์ Dagster

ตอนนี้ยังดีพลอยแค่งานเท่านั้น หากต้องการรันเวิร์กโฟลว์ ไปที่แท็บ “Runs” แล้วคลิกปุ่ม “Launch a new run” 

การรันควรเสร็จสมบูรณ์ หากต้องการดูล็อก ให้คลิกที่ ID ของรันที่สนใจ

ล็อกการรันของ Dagster

3. Mage AI

Mage AI เป็นเฟรมเวิร์กจัดระเบียบงานข้อมูลแบบไฮบริดและโอเพนซอร์ส คำว่าไฮบริดหมายถึงได้ทั้งความยืดหยุ่นแบบ Jupyter Notebook และการควบคุมแบบโค้ดโมดูลาร์ 

ใครๆ แม้มีพื้นฐาน Python จำกัด ก็สามารถสร้าง รัน และติดตามท่อทางข้อมูลได้ แทนการเขียนและรันไฟล์ Python โดยตรง จะสร้างโปรเจกต์ Mage AI และเปิดใช้งานในแดชบอร์ดเพื่อสร้าง รัน และจัดการท่อทางข้อมูล

เมื่อเทียบกับ Airflow, Mage AI มีอินเทอร์เฟซที่เป็นมิตรและใช้งานง่าย เหมาะอย่างยิ่งสำหรับผู้เริ่มต้นวิศวกรรมข้อมูล ออกแบบโดยคำนึงถึงการปรับขนาด รองรับข้อมูลปริมาณมากและโครงสร้าง pipeline ที่ซับซ้อนได้อย่างมีประสิทธิภาพ

เริ่มต้นใช้งาน Mage AI

การเริ่มต้น Mage AI ค่อนข้างง่าย เพียงติดตั้งแพ็กเกจ Python ของ Mage AI

$ pip install mage-ai

และสตาร์ตโปรเจกต์ Mage AI 

$ mage start mage_ai_etl 

คำสั่งด้านบนจะเริ่มเว็บเซิร์ฟเวอร์ ดังที่กล่าวไป การแก้ไขโค้ด การรันงาน และการติดตามงานทำผ่าน UI ของ Mage AI ทั้งหมด

หน้าจอ Mage AI

คลิก “+ New pipeline” เพื่อสร้าง ETL pipeline แรก ตั้งชื่อว่า “simple_etl”

สร้าง pipeline ใหม่ใน Mage AI

จากนั้นอินเทอร์เฟซจะให้เพิ่มโมดูลเพื่อเริ่มเขียนโค้ด เลือกโมดูล “Data Loader” และเขียนโค้ด Python ต่อไปนี้ 

ที่นี่ประกาศฟังก์ชัน create_sample_csv() ซึ่งเป็นขั้นตอนแรกของ pipeline ใช้ตัวตกแต่ง @data_loader ของ Mage AI และนิยามฟังก์ชันทดสอบ test_output() เพื่อตรวจว่ามีเอาต์พุต ช่วยจัดการการพึ่งพางาน

import io
import pandas as pd

if 'data_loader' not in globals():
    from mage_ai.data_preparation.decorators import data_loader
if 'test' not in globals():
    from mage_ai.data_preparation.decorators import test

@data_loader
def create_sample_csv() -> pd.DataFrame:
    """
    Create a sample CSV file with duplicates and missing values
    """
    csv_data = """
category,product,quantity,price
Electronics,Laptop,5,1000
Electronics,Smartphone,10,500
Clothing,T-shirt,50,20
Clothing,Jeans,30,50
Books,Novel,100,15
Books,Textbook,20,80
Electronics,Laptop,5,1000
Clothing,T-shirt,,20
Electronics,Tablet,,300
Books,Magazine,25,
"""
    return pd.read_csv(io.StringIO(csv_data.strip()))
   
@test
def test_output(df) -> None:
    """
    Template code for testing the output of the block.
    """
    assert df is not None, 'The output is undefined'

สร้างบล็อก Data Loader ใน Mage AI

ต่อไปสร้างโมดูล “Transformer” และเพิ่มฟังก์ชัน clean_data() ตามโค้ดด้านล่าง 

สามารถมองข้ามฟังก์ชัน test() ได้ เพียงเพิ่มฟังก์ชันหลักของตัวแปลงคือ clean_data() ก็พอ

import pandas as pd

if 'transformer' not in globals():
    from mage_ai.data_preparation.decorators import transformer
if 'test' not in globals():
    from mage_ai.data_preparation.decorators import test

@transformer
def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    Clean and transform the data
    """
    # Remove duplicates
    df = df.drop_duplicates()
    # Fill missing values with 0
    df = df.fillna(0)
    return df

@test
def test_output(df) -> None:
    """
    Template code for testing the output of the block.
    """
    assert df is not None, 'The output is undefined'

ในทำนองเดียวกัน สร้างโมดูล “Data Exporter” และเพิ่มโค้ดต่อไปนี้ ซึ่งประกาศฟังก์ชันโหลดข้อมูล export_data_to_csv() เพื่อบันทึกข้อมูลที่แปลงแล้วเป็นไฟล์ CSV 

import pandas as pd

if 'data_exporter' not in globals():
    from mage_ai.data_preparation.decorators import data_exporter

@data_exporter
def export_data_to_csv(df: pd.DataFrame) -> None:
    """
    Export the processed data to a CSV file
    """
    df.to_csv('output_data.csv', index=False)
    print("Data exported successfully to output_data.csv")

เพื่อรัน pipeline ไปที่แท็บ “Trigger” แล้วคลิก “Run@once”

การรัน pipeline ใน Mage AI

เพื่อดูล็อกการรัน ไปที่แท็บ “Runs” แล้วคลิกปุ่ม “Logs” ของ pipeline ล่าสุด

ล็อกการรันโฟลว์ของ Mage AI

4. Kedro

Kedro เป็นอีกหนึ่งเฟรมเวิร์กโอเพนซอร์สยอดนิยมสำหรับการจัดระเบียบงานข้อมูลที่ค่อนข้างต่างจากเครื่องมืออื่น สร้างมาเพื่อวิศวกร ML และยืมแนวคิดจากวิศวกรรมซอฟต์แวร์มาใช้กับโปรเจกต์ ML

Kedro ออกแบบมาให้เป็นโมดูลาร์สูง หมายความว่าแม้กระทั่งการส่งออกชุดข้อมูล ก็ต้องสร้าง data catalog เพื่อระบุตำแหน่งและชนิดข้อมูล ทำให้การจัดการข้อมูลตลอด pipeline เป็นมาตรฐานและมีประสิทธิภาพ

เพื่อเข้าใจว่า Kedro อยู่ตรงไหนในระบบนิเวศ ML สามารถสำรวจเครื่องมือ MLOps ต่างๆ ได้จากบทความ 25 เครื่องมือ MLOps ที่ควรรู้ในปี 2024.

เมื่อเทียบกับ Airflow, API ของ Kedro เรียบง่ายกว่าสำหรับการสร้างท่อทางข้อมูล เน้นงานวิศวกรรม ML มากกว่า และมีการจัดประเภทข้อมูลพร้อมเวอร์ชันนิ่ง

เริ่มต้นใช้งาน Kedro

การสร้างท่อทางข้อมูลด้วย Kedro ต่างออกไป เฟรมเวิร์กเป็นแบบโมดูลาร์ จึงต้องเข้าใจโครงสร้างโปรเจกต์และขั้นตอนต่างๆ เพื่อรันเวิร์กโฟลว์ได้สำเร็จ 

เริ่มจากติดตั้งแพ็กเกจ Python ของ Kedro 

$ pip install kedro

เริ่มต้นโปรเจกต์ Kedro 

$ kedro new --name=kedro_etl --tools=none --example=n 

ย้ายไปยังไดเรกทอรีโปรเจกต์ 

$ cd kedro-etl  

สร้างโฟลเดอร์ภายในโฟลเดอร์ pipelines ชื่อ data_processing.

$ mkdir -p src/kedro_etl/pipelines/data_processing  

สร้างไฟล์ Python ชื่อ kedro_pipe.py และเปิดใน IDE ที่ถนัด เช่น Visual Studio Code

$ code src/kedro_etl/pipelines/data_processing/kedro_pipe.py

สคริปต์ Python ควรมีฟังก์ชัน extract, transform และ load ซึ่งเป็นโหนดใน pipeline ในที่นี้คือฟังก์ชัน create_sample_data(), clean_data(), และ load_and_process_data()

จากนั้นเชื่อมโหนดเหล่านี้ด้วยคลาส Pipeline ของ Kedro ภายในฟังก์ชัน create_pipeline() โดยใน pipeline จะกำหนดโหนด โดยแต่ละโหนดมี inputs, outputs และ name ของโหนด 

import pandas as pd
import numpy as np
from kedro.pipeline import Pipeline, node

def create_sample_data():
    data = {
        'id': range(1, 101),
        'name': [f'Person_{i}' for i in range(1, 101)],
        'age': np.random.randint(18, 80, 100),
        'salary': np.random.randint(20000, 100000, 100),
        'missing_values': [np.nan if i % 10 == 0 else i for i in range(100)]
    }
    return pd.DataFrame(data)

def clean_data(df: pd.DataFrame):
    # Remove rows with missing values
    df_cleaned = df.dropna()

    # Convert salary to thousands
    df_cleaned['salary'] = df_cleaned['salary'] / 1000

    # Capitalize names
    df_cleaned['name'] = df_cleaned['name'].str.upper()

    return df_cleaned

def load_and_process_data(df: pd.DataFrame):
    # Calculate average salary
    avg_salary = df['salary'].mean()

    # Add a new column for salary category
    df['salary_category'] = df['salary'].apply(
        lambda x: 'High' if x > avg_salary else 'Low')

    # Calculate age groups
    df['age_group'] = pd.cut(df['age'], bins=[0, 30, 50, 100], labels=[
                             'Young', 'Middle', 'Senior'])

    print(df)
    return df

def create_pipeline(**kwargs):
    return Pipeline(
        [
            node(
                func=create_sample_data,
                inputs=None,
                outputs="raw_data",
                name="create_sample_data_node",
            ),
            node(
                func=clean_data,
                inputs="raw_data",
                outputs="cleaned_data",
                name="clean_data_node",
            ),
            node(
                func=load_and_process_data,
                inputs="cleaned_data",
                outputs="processed_data",
                name="load_and_process_data_node",
            ),
        ]
    )

หากรัน pipeline โดยไม่สร้าง data catalog ข้อมูลจะไม่ถูกส่งออก ดังนั้นไปที่ไฟล์ conf/base/catalog.yml และแก้ไขโดยระบุการตั้งค่าชุดข้อมูล

raw_data:
  type: pandas.CSVDataset
  filepath: ./data/kedro/sample_data.csv

cleaned_data:
  type: pandas.CSVDataset
  filepath: ./data/kedro/cleaned_data.csv

processed_data:
  type: pandas.CSVDataset
  filepath: ./data/kedro/processed_data.csv

ต้องเพิ่มไฟล์ Python ที่สร้างใหม่เข้าในทะเบียน pipeline ด้วย ไปที่ไฟล์ src/simple_etl/pipeline_registry.py แล้วเพิ่มโค้ดต่อไปนี้ 

"""Project pipelines."""
from __future__ import annotations

from kedro.pipeline import Pipeline
from kedro_etl.pipelines.data_processing import kedro_pipe

def register_pipelines() -> Dict[str, Pipeline]:
    data_processing_pipeline = kedro_pipe.create_pipeline()

    return {
        "__default__": data_processing_pipeline,
        "data_processing": data_processing_pipeline,
    }

รัน pipeline และดูล็อกแบบเรียลไทม์ในเทอร์มินัลด้วยคำสั่งต่อไปนี้

$ kedro run

บันทึกการรัน pipeline ของ Kedro

หลังรัน pipeline ไฟล์จะถูกบันทึกเป็น CSV ในตำแหน่งที่กำหนดไว้ใน data catalog

ไฟล์ผลลัพธ์จากการรัน Kedro pipeline

หากพบปัญหาในการรัน pipeline พิจารณาติดตั้ง Kedro พร้อมส่วนขยายทั้งหมด 

$ pip install "kedro[all]"

การแสดงภาพ Kedro

สามารถแสดงภาพและแชร์ pipeline ได้ด้วยการติดตั้งเครื่องมือ kedro-viz 

$ pip install kedro-viz

จากนั้นรันคำสั่งต่อไปนี้เพื่อแสดงภาพท่อทางข้อมูลและโหนดข้อมูลทั้งหมด พร้อมตัวเลือกติดตามการทดลองและแชร์ภาพ pipeline

$ kedro viz run

การแสดงภาพ pipeline ของ Kedro

5. Luigi

Luigi เป็นเฟรมเวิร์กโอเพนซอร์สบน Python พัฒนาโดย Spotify โดดเด่นในการจัดการงานแบตช์ระยะยาวและท่อทางข้อมูลที่ซับซ้อน เชี่ยวชาญด้านการแก้ไขการพึ่งพา การจัดการเวิร์กโฟลว์ การแสดงภาพ และการกู้คืนเมื่อเกิดความล้มเหลว จึงทรงพลังสำหรับการจัดระเบียบเวิร์กโฟลว์ข้อมูล 

เมื่อเทียบกับ Airflow, Luigi มี API แบบมินิมอล การตั้งเวลาตามปฏิทิน และมีชุมชนผู้ใช้ที่เข้มแข็งคอยช่วยแก้ปัญหาที่เกี่ยวข้องกับการจัดระเบียบท่อทางข้อมูล 

เริ่มต้นใช้งาน Luigi

การสร้างท่อทางข้อมูลด้วย Luigi ต้องเข้าใจการเขียนโปรแกรมเชิงวัตถุ เริ่มจากติดตั้งแพ็กเกจ Python ของ Luigi 

$ pip install luigi

เพื่อพัฒนา ETL pipeline แบบง่ายใน Luigi เราจะสร้างงานที่เชื่อมโยงกัน แทนการสร้างฟังก์ชัน Python เป็นงาน จะสร้างคลาส Python สำหรับแต่ละขั้นของ pipeline ได้แก่ FetchData, ProcessData และ GenerateReport โดยแต่ละคลาสจะมี 3 ฟังก์ชันคือ requires(), output() และ run() 

ฟังก์ชัน requires() และ output() จะเชื่อมงานเข้าด้วยกัน ส่วน run() จะรันโค้ดประมวลผล ท้ายสุดจะบิวด์ pipeline โดยอ้างอิงงานตัวสุดท้ายในลำดับ 

import luigi
import pandas as pd
import numpy as np

class FetchData(luigi.Task):
    def output(self):
        return luigi.LocalTarget('data/fetch_data.csv')
    
    def run(self):
        # Simulate fetching data by creating a sample CSV file
        data = {
            'column1': [1, 2, np.nan, 4],
            'column2': ['A', 'B', 'C', np.nan]
        }
        df = pd.DataFrame(data)
        df.to_csv(self.output().path, index=False)

class ProcessData(luigi.Task):
    def requires(self):
        return FetchData()
    
    def output(self):
        return luigi.LocalTarget('data/process_data.csv')
    
    def run(self):
        df = pd.read_csv(self.input().path)
        # Fill missing values
        df['column1'].fillna(df['column1'].mean(), inplace=True)
        df['column2'].fillna('B', inplace=True)
        df.to_csv(self.output().path, index=False)

class GenerateReport(luigi.Task):
    def requires(self):
        return ProcessData()
    
    def output(self):
        return luigi.LocalTarget('data/generate_report.txt')
    
    def run(self):
        df = pd.read_csv(self.input().path)
        # Simple data analysis: calculate mean of column1 and value counts of column2
        mean_column1 = df['column1'].mean()
        value_counts_column2 = df['column2'].value_counts()
        
        with self.output().open('w') as out_file:
            out_file.write(f'Mean of column1: {mean_column1}\n')
            out_file.write('Value counts of column2:\n')
            out_file.write(value_counts_column2.to_string())

if __name__ == '__main__':
    luigi.build([GenerateReport()], local_scheduler=True)

รันโค้ดข้างบนใน Jupyter Notebook หรือสร้างไฟล์ Python แล้วรันผ่านเทอร์มินัลก็ได้ 

เช่นเดียวกับ Luigi ยังสามารถเรียนรู้วิธี สร้าง ETL pipeline ด้วย Apache Airflow บทช่วยสอนครอบคลุมพื้นฐานการ extract, transform และ load ข้อมูลด้วย Apache Airflow

Luigi central planner

ต้องเริ่มต้น Luigi central planner เพื่อกำหนดตารางการรัน pipeline หรือทริกเกอร์ด้วยเหตุการณ์

สตาร์ตตัวจัดตารางด้วยคำสั่งต่อไปนี้ในเทอร์มินัล

$ luigid  

2024-06-22 13:35:18,636 luigi[25056] INFO: logging configured by default settings
2024-06-22 13:35:18,636 luigi.scheduler[25056] INFO: No prior state file exists at /var/lib/luigi-server/state.pickle. Starting with empty state
2024-06-22 13:35:18,640 luigi.server[25056] INFO: Scheduler starting up

หากต้องการรัน pipeline ให้เปิดเทอร์มินัลใหม่แล้วพิมพ์คำสั่งต่อไปนี้ คำสั่ง Luigi ต้องระบุชื่อไฟล์ Python และงานสุดท้ายที่ต้องการรัน ในที่นี้ชื่อไฟล์คือ luigi_pipe.py และงานสุดท้ายคือ GenerateReport.

$ python -m luigi --module luigi_pipe GenerateReport

หากต้องการแสดงภาพการรัน pipeline และสถานะงาน เข้าไปที่ http://localhost:8082 ในเบราว์เซอร์

เว็บ UI ของ Luigi Central Planner

จบแล้วสำหรับการพาไปรู้จัก 5 ทางเลือกที่ดีที่สุดแทน Airflow หากต้องการเจาะลึกตัวอย่างใดในบทความนี้ นี่คือแหล่งข้อมูลเพิ่มเติม:

• ซอร์สโค้ดและข้อมูลของ Prefect, Dagster และ Luigi ดูได้ที่ DataLab workspace.
• ซอร์สโค้ดและข้อมูลของ Mage AI และ Kedro ดูได้ที่ GitHub repository.

ข้อคิดส่งท้าย

ในบทช่วยสอนนี้ เราได้พูดถึงทางเลือก Airflow แบบโอเพนซอร์สและใช้ฟรีที่โดดเด่น เรียนรู้เครื่องมือจัดระเบียบงานข้อมูลแต่ละตัว พร้อมทั้งสร้างและรันท่อทาง ETL อย่างง่าย การเห็นโค้ดตัวอย่างจะช่วยตัดสินใจได้ว่าเครื่องมือใดเหมาะกับกรณีใช้งานของตน

หากเป็นผู้เริ่มต้น แนะนำให้เริ่มที่ Prefect หรือ Mage AI เพราะใช้งานง่ายและตั้งค่าง่าย แต่หากมองหาเครื่องมือขั้นสูงที่ยึดแนวปฏิบัติด้านวิศวกรรมซอฟต์แวร์ แนะนำให้สำรวจ Dagster, Kedro และ Luigi

หลังอ่านบทความนี้แล้ว ก้าวต่อไปในการเป็นวิศวกรข้อมูลคือการรับรองทักษะ เช่นเส้นทางของ DataCamp Data Engineer in Python เพื่อเรียนรู้เครื่องมืออื่นๆ และสร้างท่อทางข้อมูลแบบ end-to-end ที่ดีพลอยสู่โปรดักชันได้