Airflow

HookLineageCollector와 AIP-60 Asset

Airflow는 태스크 사이뿐 아니라 태스크 안에서 쓰이는 훅으로부터도 데이터 lineage를 추적할 수 있는 기능을 제공한다 — 다만 이 lineage 지원은 매우 실험적이며 앞으로 바뀔 수 있다. 이 기능은 파이프라인 전체에서 데이터가 어떻게 흐르는지 이해하는 데 도움을 준다.

전역 ``HookLineageCollector`` 인스턴스가 lineage 정보를 모으는 중앙 허브 역할을 한다. 훅은 자신이 다루는 asset에 대한 세부 정보를 이 수집기에 보낼 수 있고, 수집기는 이 데이터를 이용해 AIP-60 호환 Asset(asset을 기술하는 표준 포맷)을 구성한다. 훅은 asset과 무관한 임의의 데이터도 수집기에 보낼 수 있다.

예를 들어 커스텀 훅의 ``run`` 메서드 안에서 ``get_hook_lineage_collector()``로 수집기를 얻은 뒤, ``collector.add_input_asset(self, asset_kwargs={...})``와 ``add_output_asset(...)``으로 입력·출력 asset을 기록하고, ``add_extra(self, key=..., value=...)``로 임의의 부가 데이터(예: 외부 시스템의 job id)를 함께 기록할 수 있다.

수집된 lineage 데이터는 Airflow 플러그인에 등록된 ``HookLineageReader`` 인스턴스로 읽을 수 있다 — 예를 들어 ``HookLineageReader``를 상속해 ``get_inputs()``를 오버라이드한 커스텀 리더를 만들고, ``AirflowPlugin`` 서브클래스의 ``hook_lineage_readers`` 리스트에 등록하면 된다. 만약 어떤 ``HookLineageReader``도 등록되어 있지 않다면, 기본값인 ``NoOpCollector``가 대신 쓰인다 — 이 수집기는 AIP-60 호환 Asset을 만들지도, lineage 정보를 수집하지도 않는다.

수집 한도 — max_assets_per_collector, max_extras_per_collector

수집기는 입력 asset을 ``[lineage] max_assets_per_collector`` 개수까지, 출력 asset도 (입력과 별도로 카운트해) 같은 개수까지, 그리고 부가 메타데이터 항목을 ``[lineage] max_extras_per_collector`` 개수까지만 보관한다. 한도를 넘어서는 항목은 버려지므로, 긴 루프 안에서 반복 호출되는 훅은 자신의 한도에 도달한 이후로는 더 이상 lineage에 기여하지 못하게 된다. 어떤 한도를 0으로 설정하면 그 종류의 수집 자체를 완전히 비활성화한다.