Metadata-Version: 2.4
Name: datahub-airflow-plugin-lite
Version: 0.1.2
Summary: Dependency-free DataHub listener plugin for Apache Airflow
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: acryl-datahub[datahub-rest]; extra == "dev"

# datahub-airflow-plugin-lite

Airflow 실행 메타데이터와 lineage를 DataHub로 보내는 **런타임 의존성 0개** 플러그인입니다.
표준 라이브러리와 이미 설치된 Airflow만 사용하므로 MWAA 등 Airflow constraints 파일과 충돌하지 않습니다.

| 항목        | 지원 범위             | 검증한 조합                                        |
| ----------- | --------------------- | -------------------------------------------------- |
| Python      | 3.9 이상              | 3.9.6, 3.12.13                                     |
| Airflow     | 2.9.1 이상 (3.x 포함) | 2.9.1, 2.11.2, 3.1.0                               |
| DataHub GMS | 1.4.0.3 이상          | URN/aspect를 acryl-datahub 1.4.0.3 SDK 출력과 대조 |

## 설치

```bash
pip install datahub-airflow-plugin-lite-0.1.2-py3-none-any.whl
```

공식 플러그인(`acryl-datahub-airflow-plugin`)과 동시에 설치하지 마세요. 두 리스너가 같은 엔티티를 중복 전송합니다.

## 설정

`airflow.cfg`의 `[datahub]` 섹션(또는 `AIRFLOW__DATAHUB__*` 환경변수)을 읽습니다. 키 이름은 공식 플러그인과 같습니다.

```ini
[datahub]
# GMS 주소/토큰을 직접 지정하거나, 생략하면 conn_id 커넥션(host=GMS URL, password=토큰)을 사용
gms_url = https://datahub-gms.example.com
token = <personal access token>
conn_id = datahub_rest_default
cluster = prod
platform_instance =
capture_ownership_info = true
capture_ownership_as_group = false
capture_tags_info = true
capture_executions = true
materialize_iolets = true
capture_airflow_assets = true
enable_datajob_lineage = true
enable_openlineage_facets = true
openlineage_timeout_sec = 5
supplement_with_sql_parser = true
athena_include_result_location = false
dag_filter_str = {"allow": [".*"], "deny": []}
timeout_sec = 10
retries = 2
ssl_verify = true
```

운영 스위치도 공식 플러그인과 같습니다.

- `AIRFLOW_VAR_DATAHUB_AIRFLOW_PLUGIN_DISABLE_LISTENER=true`: 리스너 즉시 중지 (kill switch)
- `DATAHUB_AIRFLOW_PLUGIN_RUN_IN_THREAD`, `DATAHUB_AIRFLOW_PLUGIN_RUN_IN_THREAD_TIMEOUT`: 백그라운드 스레드 실행 여부와 대기 시간

## 동작

| Airflow 이벤트   | 전송 내용                                                                                                                     |
| ---------------- | ----------------------------------------------------------------------------------------------------------------------------- |
| DAG run 시작     | DataFlow (`dataFlowInfo`, `status`, `browsePathsV2`, `ownership`, `globalTags`, `dataPlatformInstance`), 태스크·태그 `status` |
| 태스크 시작      | DataJob (`dataJobInfo`, `status`, `ownership`, `globalTags`), 실행 기록 (`dataProcessInstance*`, STARTED)                     |
| 태스크 성공/실패 | DataJob + `dataJobInputOutput`(lineage), 실행 결과(COMPLETE, SUCCESS/FAILURE), 실행 입출력, 데이터셋 `status`                 |

전송은 GMS의 `POST /aspects?action=ingestProposal` 하나만 사용합니다.
모든 처리는 데몬 스레드에서 실행되고, 예외는 로그로만 남기므로 태스크 실패로 이어지지 않습니다.

### Lineage 소스

1. `inlets`/`outlets`의 `Dataset`/`Urn` 객체 또는 `urn:li:...` 문자열
   - `datahub_airflow_plugin_lite.entities.Dataset`/`Urn`
   - 공식 플러그인의 `datahub_airflow_plugin.entities` 객체도 `urn` 속성이 있어 그대로 인식합니다.
2. Airflow 네이티브 Dataset(2.x)/Asset(3.x). URI 스킴으로 플랫폼을 판별합니다.
3. **OpenLineage provider (기본 소스):** operator의 `get_openlineage_facets_on_complete/start()` 결과를 사용합니다. 몽키패치는 하지 않습니다.
   - **정규화:** provider 버전마다 다른 결과를 DataHub ingestion URN으로 맞춥니다(`ol_normalize.py`).
     - Athena: `AwsDataCatalog.` 접두어를 제거하고 소문자로 바꿉니다.
     - S3: 버킷과 키를 합쳐 `make_s3_urn`과 같은 규칙을 적용합니다.
     - Snowflake: 소문자로 바꿉니다.
   - **버전 의존 값 제거:** 버전마다 다르게 나오는 Athena 쿼리 결과 위치는 제외합니다.
   - **시간 제한:** OpenLineage 호출은 `openlineage_timeout_sec`(기본 5초)까지만 기다립니다.
4. **내장 SQL 파서 (보완):** OpenLineage가 버전에 따라 놓친 테이블을 채웁니다. 예: provider 2.x의 CTE 입력 누락, amazon 8.x의 Athena 테이블 누락. 이 단계는 OpenLineage 결과를 지우지 않습니다.
   - DB에 접속하지 않고 SQL 문과 Connection 정보(conn_type, 기본 DB/스키마)만 사용합니다.

lineage 수집이 실패해서 결과가 불완전할 수 있으면 `dataJobInputOutput`을 보내지 않습니다. 이미 저장된 lineage를 덮어쓰지 않기 위해서입니다. 예: OpenLineage가 실패했고 SQL도 분석할 수 없는 경우.
DataJob 정보와 실행 기록은 lineage보다 먼저 전송하므로, lineage 수집 결과와 관계없이 항상 기록됩니다.

provider 1.7.0, 2.11.0, 2.20.2에서 같은 lineage가 저장되는지 비교한 결과는 [tests/compat/README.md](tests/compat/README.md)에 있습니다.

## 공식 플러그인과의 차이

URN 생성 규칙(DataFlow, DataJob, Dataset, DataProcessInstance)은 공식 플러그인과 같습니다. 따라서 공식 플러그인에서 전환해도 엔티티가 중복 생성되지 않습니다.
`tests/unit/test_sdk_parity.py`가 이를 검증합니다.

| 기능                                                            | 공식              | lite                                           |
| --------------------------------------------------------------- | ----------------- | ---------------------------------------------- |
| DataFlow/DataJob/실행 이력, 소유자, 태그                        | O                 | O                                              |
| inlets/outlets, Airflow Dataset/Asset lineage                   | O                 | O                                              |
| OpenLineage 기반 테이블 lineage                                 | O (SQL 파서 패치) | O (패치 없이 결과 정규화 + 내장 SQL 파서 보완) |
| 컬럼 단위 lineage (DataHub SQL 파서)                            | O                 | X                                              |
| Kafka/File emitter, 다중 커넥션                                 | O                 | X (REST만)                                     |
| `Datahub_Cleanup` DAG, DatahubEmitterOperator, assertion sensor | O                 | X                                              |
| Jinja 템플릿 렌더링 후 속성 수집                                | O                 | X (원본 값 사용)                               |

## 개발

```bash
pip install -e ".[dev]"
pytest
```

`test_sdk_parity.py`는 `acryl-datahub`가 설치된 경우에만 실행됩니다. `test_listener.py`는 설치된 Airflow의 실제 listener manager(pluggy)에 등록해 훅 시그니처 호환성을 검증합니다.
