【10】 NIFI를 이용한 DBMS JSON 데이터 전처리 (with. Ndap Workflow Query) : kt NexR 기술블로그 ‘21 테크리포트

Ndap Workflow는 oozie Workflow 기반으로 만들어진 작업의 순서를 정하는 기능을 얘기 합니다. 순서가 있어 시작점과 끝점이 존재 합니다. 쿼리, SSH를 이용한 shell 명령 등 다양한 작업이 가능 합니다.

(Example)

NDAP의 메타 데이터는 Mysql을 통해 DB로 저장한다. 워크플로우 정보 또한 NDAP 메타DB workflow 테이블에 저장된다.

(example)

jsonDefinition 컬럼의 row 구조는 다음과 같다. {Name: workflow name, Position:{x:000:y000}, Type : HIVE,SSH,JDBC … Query : alter….. Monitoring: Key… }

NIFI executeSQL을 사용하여 데이터를 추출한다.

(example)

SplitAvro 를 이용하여 한 row씩 데이터를 추출한다. List queue

List queue 결과

Split 결과

ConvertAvroToJson 프로세서를 사용해 Avro->Json포맷으로변경

EvaluateJsonPath프로세서를 사용하여 원하는 key를 변수 선언한다. (name이란 이름으로 변수 선언을 하고 $.name하면 실제 json의 name 값을 가져온다. NIFI 표현식 참고)

(jsonDefinition과 name컬럼을 변수화 처리 하였음)

잘 나오는지 queue의 attribute 값을 확인한다.

(변수화 선언 후 attribute 값)

attribute를 보면 jsonDefinition과 name을 변수로 잘 선언했고 값도 잘 가져왔음을 확인할 수 있다. 그럼 변수도 확인했으니 본격적인 json 추출 작업을 진행해야 한다.. 첫번째로 [{ "jsonDefinition" : "{\"nodes\":[{.........}]}" "name" : "workflow_name" }] 부분에서 name 을 자르자. 왜냐면 어차피 변수처리해서 attribute에 남아있고 jsonDefinition이 2depth이상의 json이기 때문이다. Replacetext 프로세서를 이용해 값을 치환한다.

(Replacetext 프로세서를 이용해 전체값 -> jsondefinition값 치환)

(치환 후 attribute 값)

위와같이 evaluatejsonpath 프로세서를 이용해 nodes값 선언

(attribute 값)

같은 규칙을 가진 다중 Json구문은 splitJson 프로세서를 이용한다.

Evaluatejsonpath 프로세서를 이용해 query를 변수로 담는다.

(attribute로 query를 변수로 담은 장면)

위와같이 replacetext를 통해 전체 텍스트를 query로 치환한다.

(view 값을 통해 쿼리 확인.)

Mergecontent 프로세서를 이용해 처음에 선언한 name변수를 기준으로 query를 합친다.

(merge 후)

Merge 후 적재

결과확인

전체 플로우

기존 운영팀에서 가지고 있던 쿼리추출 실행파일 같은 경우엔 쿼리만 추출이 가능하다. 하지만 NIFI를 이용하면 내가 원하는 데이터를 기준으로 추출 및 정렬, 병합이 가능할 뿐더러 꼭 NDAP worflow가 아니더라도 데이터 수집 및 전처리시 막강한 확장성과 자유도를 가진다.

【10】 NIFI를 이용한 DBMS JSON 데이터 전처리 (with. Ndap Workflow Query) : kt NexR 기술블로그 ‘21 테크리포트