從 0 開始貢獻 Apache Airflow
#

為什麽選擇 Apache Airflow ?
#

想要從 Apache Foundation 的 Top Level Project 開始貢獻
看到 Apache Airflow 有 38.6k 的 star
在 Data Team 看也發現 Airflow 確實也是 Data Engineering 領域的重要工具
目前也對 Python 最為熟悉

背景
#

在實際貢獻 Apache Airflow 之前
我其實只有在跟 Data Engineering 相關的部門實習到當時只有 3 個月多
負責的 task 其實連 DAG 都沒有寫到
主要都還是負責 General Backend 的工作

貢獻統計
#

先講一下到目前的貢獻統計

總 PR 數: 50+
#

Link to Total Merged PR

在 GitHub 的貢獻排名 ( 專案從開始至今 ): Rank 72
#

Link to Contribution Graph on GitHub

在 OSS Rank 看到的貢獻排名 ( 有根據近期貢獻加權 ): Rank 29
#

Link to OSS Rank

第一個 PR
#

從去年 2024 年 10 月初開始正式開始貢獻 Apache Airflow
當是是看到 Fix PythonOperator DAG error when DAG has hyphen in name

有被標記為 good first issue
就嘗試 trace 一下看看，發現應該只需要改一行程式碼
就決定試看看

源來適你
#

源來適你是一個實際貢獻開源台灣的的組織

這邊是關於 源來適你 更深入的介紹 Kafka Community Spotlight: TAIWAN 🇹🇼 by Stanislav’s Big Data Stream 除了 #kafka 之外，也包括 #airflow頻道

因為第一個 Issue 剛好跟 DAG 有關
按照 Doc 去在 Breeze Container 去 reproduce 時步驟有點問題
向 Committer @Lee-W 大大請教

應該算是李維大大的 mentee xD (?)
之後有遇到問題或是 PR 需要 Review 、加 Label 都會請他幫忙！

李維大大的 Blog 貢獻 Airflow 101: 姑且算是個 mentor(?)…吧？

第一個 PR Merged
#

發出第一個 Apache Airflow 的 PR: Fix PythonOperator DAG error when DAG has hyphen in name #42902
剛好讓原本不認得的隔壁部門同事 @josix 幫忙 review 到 xD

雖然主要改的只有一行程式碼
但中間其實有超過 20 個 comment 來回迭代修正
也讓我知道開源其實沒有那麼剪改一行 code 這麼簡單

尤其是在 Unit Test 的部分
之前主要都寫 integration Test 比較沒有寫到 mock 的經驗

管理 Tasks 的方式
#

前期使用 HackMD 以 Markdown 來簡單紀錄最近看到可以研究的 Issue

使用 HackMD 紀錄的 Issue List

目前都直接使用 GitHub Projects 的 Kanban 來管理
因為同時可能有 2-3 個 Issue 正在解
又會有些是在等 Code Review 的階段
還有逛 Issue List 看到有機會做的 Issue 可以放在 Backlog

使用 GitHub Projects 管理的 PR List

前 50 個 PR
#

以下的 AIP-XX 都是指 Airflow Improvement Proposal 的其中一個提案

AIP-84: Modern REST API
#

剛好去年 10 月開出很多 AIP-84 的 Issues
主要是把 legacy API ( Flask 寫的 API ) migrate 到 FastAPI 的 API
因為當時最熟的就是 FastAPI 所以總共接了快 10 個 API Migration

在做這些 API Migration 的過程中也多少學到不少 Airflow 的架構
還有寫 test 常用到的 pytest fixture 如 dag_maker 、 dag_bag 、 create_dag_run、create_task_instances 等等

重構 Parameter System
#

Context
在 FastAPI 架構下，每個 filter ( 竟更精確來說是 query parameter) 都會繼承 BaseParam
當 API 的 filters 很多時，透過 BaseParam 的架構可以讓 router 層比較乾淨

BaseParam 的定義如下

67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class BaseParam(Generic[T], ABC):
    """Base class for filters."""

    def __init__(self, value: T | None = None, skip_none: bool = True) -> None:
        self.value = value
        self.attribute: ColumnElement | None = None
        self.skip_none = skip_none

    @abstractmethod
    def to_orm(self, select: Select) -> Select:
        pass

    def set_value(self, value: T | None) -> Self:
        self.value = value
        return self

    @abstractmethod
    def depends(self, *args: Any, **kwargs: Any) -> Self:
        pass

Problem
但隨著越來越多 API 被 Migrate 到 FastAPI 的架構
每個 API 都在 common/parameters.py 的 module 加上繼承 BaseParam 的 class
有 n 個 entity 的 API 就會多出 n 個 class

所以應該要有一個通用的 Factory Pattern 並對 FastAPI 所需的 typing 綁定來產生這些 class
經過這個重構的 PR 應該 50+ 個 API 都有利用到 filter_param_factory

304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def filter_param_factory(
    attribute: ColumnElement,
    _type: type,
    filter_option: FilterOptionEnum = FilterOptionEnum.EQUAL,
    filter_name: str | None = None,
    default_value: T | None = None,
    default_factory: Callable[[], T | None] | None = None,
    skip_none: bool = True,
    transform_callable: Callable[[T | None], Any] | None = None,
) -> Callable[[T | None], FilterParam[T | None]]:
    # if filter_name is not provided, use the attribute name as the default
    filter_name = filter_name or attribute.name
    # can only set either default_value or default_factory
    query = (
        Query(alias=filter_name, default_factory=default_factory)
        if default_factory is not None
        else Query(alias=filter_name, default=default_value)
    )

    def depends_filter(value: T | None = query) -> FilterParam[T | None]:
        if transform_callable:
            value = transform_callable(value)
        return FilterParam(attribute, value, filter_option, skip_none)

    # add type hint to value at runtime
    depends_filter.__annotations__["value"] = _type

    return depends_filter