NiFi PutHDFS 적재 후 Airflow DAG 호출 구성

1. 전체 구성

HDFS staging 적재부터 Airflow 호출까지 NiFi 플로우의 큰 흐름과 각 단계의 역할 정리

1.1 플로우 흐름

PutHDFS(staging) → conf 파일 내용 생성 → PutHDFS(conf 폴더) → InvokeHTTP(Airflow)

1.2 단계별 역할

• PutHDFS(staging) : 데이터 파일을 staging 경로에 적재함. 성공 시 absolute.hdfs.path 속성이 FlowFile에 생성됨
• PutHDFS(conf 폴더) : result_full_path가 담긴 conf 파일을 저장함. DAG가 이 파일을 읽어 staging 파일의 위치를 확인하는 구조임
• InvokeHTTP : Airflow dagRuns API에 {"conf": {...}} 본문을 POST함. DAG 안에서는 dag_run.conf로 읽힘

staging 데이터와 conf 파일이 모두 저장된 뒤에 Airflow를 호출해야 하므로, 세 단계는 success 관계로 직렬 연결하는 것이 가장 단순하다.

---

2. Airflow 호출 본문과 conf

InvokeHTTP로 보내는 JSON 본문의 구조와 `${...}` 표기의 정체 정리

2.1 Expression Language

• ${...} 표기 : NiFi Expression Language이며, FlowFile 속성값으로 치환됨
• 속성이 없는 경우 : 빈 문자열로 치환되므로 값이 비어 있으면 앞 단계에서 속성이 만들어졌는지 확인 필요
• 본문 구성 위치 : InvokeHTTP는 FlowFile 내용을 그대로 전송하므로, JSON 본문은 앞 단계의 ReplaceText에서 구성함

2.2 dagRuns API의 conf

• 호출 경로 : POST /api/v1/dags/{dag_id}/dagRuns
• conf 키 : 본문의 conf 객체가 DAG 실행 파라미터가 되며, DAG 안에서 dag_run.conf로 접근함
• HDFS conf 폴더의 파일 : Airflow API와는 별개로, 해당 파이프라인(Spark 포함)이 정한 규약임. 파일이 없으면 Spark 단계에서 실패하는 구조

{
  "conf": {
    "sys_id": "${abd}",
    "result_full_path": "${staging_path}"
  }
}

conf는 NiFi 문법이 아니라 Airflow가 받는 본문의 키 이름이다. PutHDFS의 Hadoop Configuration Resources 속성과는 무관하다.

---

3. 경로 속성 처리

PutHDFS가 생성하는 속성과 덮어쓰기 문제, 경로 접두어 처리 정리

3.1 PutHDFS가 생성하는 속성

• absolute.hdfs.path : PutHDFS 성공 후 FlowFile에 추가되는 저장 디렉터리 경로
• filename : 저장된 파일명. UpdateAttribute에서 지정한 값이 유지됨

3.2 속성 덮어쓰기

두 번째 PutHDFS를 거치면 absolute.hdfs.path가 conf 폴더 경로로 바뀌고, conf 파일명을 지정하려고 filename을 바꾸면 staging 파일명이 사라짐. 첫 PutHDFS 직후 별도 속성에 담아 두는 방식으로 해결함.

UpdateAttribute (staging PutHDFS 직후)
  staging_path = ${absolute.hdfs.path}/${filename}

• 이후 단계 : conf 파일 내용과 Airflow 본문 모두 ${staging_path}를 참조
• conf 파일 내용 구성 : ReplaceText의 Evaluation Mode를 Entire text로 두어 기존 내용을 통째로 교체

3.3 hdfs:// 접두어

• 접두어 포함 여부 : absolute.hdfs.path에 hdfs://가 붙는지는 환경과 버전에 따라 다름
• DAG가 포함 경로를 기대하는 경우 : hdfs://네임노드/ 형태로 맞춰 staging_path를 구성해야 함

실제 속성값을 한 번 확인하고, DAG가 받는 형식에 맞춰 staging_path를 만드는 것이 안전하다.

---

4. 실행 순서와 응답 처리

호출 순서 보장과 InvokeHTTP 응답의 의미 정리

4.1 순서 보장

• 직렬 연결 : staging PutHDFS → conf PutHDFS → InvokeHTTP 순으로 success 연결. conf 파일 내용은 ReplaceText로 새로 만들어야 하므로 중간에 변환 단계가 필요함
• 분기 후 합류 : FlowFile을 복제해 각각 적재하는 방식은 Wait/Notify나 MergeContent가 필요해 복잡해짐
• 먼저 호출한 경우 : staging 데이터나 conf 파일이 없어 Spark가 파일을 찾지 못하고 실패함

4.2 응답과 운영 설정

• 200 응답의 의미 : DAG 실행 등록 완료를 뜻할 뿐, Spark 작업 성공을 뜻하지 않음. 성공 여부는 Airflow 쪽 상태로 별도 확인 필요
• PutHDFS 충돌 설정 : 같은 이름 파일이 있을 때 replace, ignore, fail 중 선택. 재실행 시 동작이 달라지므로 사전에 결정
• 인증 : Airflow API에 Basic 인증이나 토큰이 필요하면 InvokeHTTP의 Username/Password 또는 Authorization 헤더로 설정

---

요약

구분	내용
흐름	PutHDFS(staging) → PutHDFS(conf) → InvokeHTTP(Airflow)
conf의 정체	Airflow dagRuns 본문의 키 이름. NiFi 문법 아님
${...}	FlowFile 속성값 치환 (Expression Language)
경로 속성	absolute.hdfs.path, filename은 PutHDFS마다 덮어쓰이므로 staging_path로 보존
접두어	hdfs:// 포함 여부는 DAG 기대 형식에 맞춰 구성
응답	200은 실행 등록이며 작업 성공이 아님

staging 적재, conf 파일 적재, Airflow 호출 순서를 지키고 경로 속성은 덮어쓰이기 전에 따로 저장해 두는 것이 핵심이다.