데이터 적재 연습하기 04 _ 데이터베이스 적재된 데이터와 FTP서버의 파일데이터

 

 

 

📖 데이터 적재 / 변환 시나리오 _ 데이터베이스 적재된 데이터와 FTP서버의 파일데이터 

 

 

데이터베이스에 적재된 데이터와 FTP서버의 파일데이터를 합쳐 DB의 nifi 스키마의 company_20230216 테이블에 적재

적재한 데이터를 cvs 형식 변경하여 FTP서버에 파일 지정한다.

(DB 데이터 + 파일 데이터)

 

 

 

1. postgres 데이터베이스의 nifi 스키마의 ovs_expns_kor_entprs 테이블에서 나라명이 '라오스'인 데이터만 조회

2. 조회한 데이터를 AVRO 에서 JSON으로 레코드 형식 변환

3. 변환된 JSON 데이터를 한줄씩 분리 ( json 분리 기준은 $.* 로 )

4. 분리한 JSON 데이터에서 country_nm, exp_plc_nm, region_se 항목을 플로우파일 속성으로 등록

5. FTP 서버 /nifi_data/raw_dataset 디렉토리에서 load_20230216.json 데이터 불러오기

6. 불러온 JSON 데이터에서 entprs_eng_nm, entprs_kor_nm, exp_yy 항목을 플로우파일 속성으로 등록

7. postgres 데이터베이스의 nifi 스키마의 company_20230216 테이블에 4, 6번에서 등록한 속성들의 값으로 컬럼명과 매핑하여 데이터 적재
   적재일자(load_de) 컬럼에는 오늘날짜(YYYYMMDD) 값 적재

※ 컬럼정보 (region_se, country_nm, exp_plc_nm, entprs_kor_nm, entprs_eng_nm, exp_yy, load_de)

8. 적재완료 후 하나의 JSON 데이터로 합치기

9. 합친 데이터를 JSON 에서 CSV로 레코드 형식 변환

10. 파일명을 이니셜_20230216.csv 로 변경

11. FTP 서버 /nifi_data/result_dataset 디렉토리에 이니셜 폴더 하위에 파일 적재

 

 

 

 

 

 

 

시나리오 실행 Flow

 

 

 

 

 

 

 

- ExecuteSQL Processor

 

스케쥴을 10 min으로 설정

기본 '0 sec'로 실행 시켰더니 데이터가 빠르게 계속 조회되어서 시간을 지정해두었다

 

 

Database Connection Pooling Service : 연결할 데이터베이스 컨트롤러를 설정한다.

SQL select query : 데이터를 조회할 SQL query를 작성 해준다.

select * from nifi.ovs_expns_kor_entprs
where country_nm = '라오스'

ExecuteSQL에서 country_nm='라오스'인 데이터를 조회해서 flowfile 생성

 

 

 

 

 

- ConvertAvroToJSON Processor

 

조회한 Avro 데이터를 json 형식으로 변경한다

 

 

ConvertRecord 프로세스를 활용해서 파일 형식을 변환 할 수있다.

(Avro에 대한 컨트롤러를 생성 해야함.)

 

 

 

 

 

 

 

- SplitJson Processor

 

변환된 Json 데이터를 json형식으로 ('$.*') 구분하여 데이터를 분리

 

 

 

 

 

- EvaluateJsonPath Processor

 

 

분리한 JSON 데이터에서 country_nm, exp_plc_nm, region_se 항목을 플로우파일 속성으로 등록

 

데이터의 속성을 생성하여 flow안에 데이터 덮어씌어지거나 사라져도 데이터를 사용할 수 있게 해준다(?!)

 

 

 

 

 

 

 

- FetchFTP Processor

FTP서버 /nifi_data/raw_dataset 디렉토리에서 load_20230216.json 파일 데이터 불러오기

 

 

 

 

 

- EvaluateJsonPath Processor

 

불러온 데이터 EvaluateJsonPath 프로세스로 entprs_eng_nm, entprs_kor_nm, exp_yy 항목 속성 생성

 

 

 

 

 

 

 

- UpdateAttribute Processor

 

postgres 데이터베이스의 nifi.company_20230216 테이블에 속성 값으로, 테이블의 컬럼과 매칭하여 데이터를 적재한다
이때, 적재일자 load_de 컬럼에는 오늘날짜(YYYYMMDD) 형식으로 값 적재되도록 속성을 생성해준다.

 

정규식 ' ${now():format('yyyyMMdd') ' 을 사용해줄 수 있다.

 

 

 

 

 

 

 

- PutSQL Processor

 

JDBC Connection Pool : 연결할 데이터베이스 컨트롤러 선택

SQL Statement : 실행할 INSERT or UPDATE 쿼리문을 작성한다.

Support Fragmented Transactions : 

true이면 단일 트랜잭션을 처리하고, false이면 여러개의 flowfile 다중 트랜잭션을 처리한다

 

PutSQL 프로세스로 데이터베이스에 insert 쿼리를 작성하여 데이터를 적재한다.

 

insert into nifi.company_20230216(region_se,country_nm,exp_plc_nm, entprs_kor_nm,entprs_eng_nm,exp_yy,load_de)
values('${region}','${country}','${exp_plc}','${entprs_kor}','${entprs_eng}','${exp}','${time}')

 

데이터가 적재 된 모습이다.

 

 

 

 

 

 

 

 

 

- MergeRecord Processor

 

데이터베이스에 데이터가 적재되었는지 확인하고 
MergeRecord 프로세스를 활용하여 플로우의 데이터파일의 내용을 하나의 json 파일로 합친다.

 

 

 

 

 

 

 

- ConvertRecord Processor

 

데이터 적재 후  json파일의 형식을 하나의 cvs로 형식을 변환한다

 

 

 

 

 

 

- UpdateAttribute Processor

 

UpdateAttribute 프로세스로 파일명을 br_20230216.csv 로 변경

time 속성을 사용하여 오늘 날짜가 들어가도록 해보았다!

속성 활용!!

 

 

 

 

 

 

 

- PutFTP Processor

 

 

FTP 서버 /nifi_data/result_dataset 위치에 파일이 저장되도록 한다.

 

 

 

 

 

 

 

 

 

 

파일이 지정된 경로에 저장 되었다.

 

 

 

 

 

지정된 경로에 저장된 파일을 열어보면 이렇게 데이터가 나온다.

3개 컬럼에 대한 데이터가 나오는게 맞다

 

FTP서버에서 가져온 flow파일에 대한 컬럼은 3가지 이므로 데이터베이스 데이터를 적재했다 해도

FTP서버에 올려지는 데이터는  entprs_eng_nm, entprs_kor_nm, exp_yy 항목만 나오는것!