데이터 적재/변환 연습하기 05_정규식을 사용하여 속성 생성 , 파일 해제 후 속성으로 데이터 수정하고 DB 적재된 데이터 Delete 후 INSERT 하여 데이터 적재 후 파일로 압축하여 SFTP 서버에 적재

 

 

 

 

📖 데이터 적재 / 변환 시나리오

 

 

 

1. FTP 서버 (172.30.1.3) /nifi_data/raw_dataset/ 디렉토리에서 business_district_data.zip 파일 수집 (원본 유지)

2. 오늘날짜(yyyymmdd) 값이 들어간 time 이름의 속성을 생성 (값을 숫자사용하지 않고 Expression Language 를 사용하여 생성)

3. 수집한 zip 파일 압축해제

4. 파일 명에 오늘 날짜값이 들어간 파일들만 분리

5. 데이터 건수 측정

6. 데이터 수정
1) 파일의 컬럼명을 아래의 값으로 변경

상가업소번호 house_no
상호명 cmpny_nm
지점명 point_nm
상권업종대분류코드 industry_lgcls_cd
상권업종대분류명 industry_lgcls_nm
상권업종중분류코드 industry_mdcls_cd
상권업종중분류명 industry_mdcls_nm
상권업종소분류코드 industry_smcls_cd
상권업종소분류명 industry_smcls_nm
표준산업분류코드 std_cls_cd
표준산업분류명 std_cls_nm
시도코드 sd_cd
시도명 sd_nm
시군구코드 sgg_cd
시군구명 sgg_nm
행정동코드 hd_cd
행정동명 hd_nm
법정동코드 bd_cd
법정동명 bd_nm
지번코드 addr_cd
대지구분코드 site_cd
대지구분명 site_se_nm
지번본번지 addr_main_no
지번부번지 addr_sub_no
지번주소 addr_data
도로명코드 road_cd
도로명 road_nm
건물본번지 bud_main_no
건물부번지 bud_sub_no
건물관리번호 bud_mng_no
건물명 bud_nm
도로명주소 bud_addr
구우편번호 old_zip_cd
신우편번호 new_zip_cd
동정보 dong_info
층정보 flor_info
호정보 room_info
경도 logtd
위도 lattd


2) 파일의 컬럼 중 경도, 위도컬럼의 값에서 . 을 공백으로 대체
3) 5에서 측정한 데이터 건수의 값이 들어가는 data_co 컬럼 생성
4) 2에서 생성한 time 속성의 값이 들어가는 load_de 컬럼 생성



9. postgresql DB nifi 스키마 내의 이니셜_business_dist_data 테이블의 데이터를 DELETE

10. postgresql DB nifi 스키마 내의 이니셜_business_dist_data 테이블에 수집한 데이터 INSERT

11. 적재 완료된 데이터들을 zip 포맷으로 압축

12. 압축된 파일의 파일명을 이니셜_오늘날짜.zip 로 변경

10. 파일을 SFTP서버의 /data01/result_data/ 디렉토리 하위에 이니셜로 폴더를 만들어 적재

 

 

 

 

 

 

 

 

시나리오 Flow

 

 

 

 

 

 

 

 

 

 

 

GetFTP 프로세스로 FTP 서버 (172.30.1.3) /nifi_data/raw_dataset/ 디렉토리에서 business_district_data.zip 파일 수집

(원본 유지)

 

 

 

 

 

 

 

UpdateAttribute 프로세스로 속성 생성

오늘날짜(yyyymmdd) 값이 들어간 time 이름의 속성을 생성

(값을 숫자사용하지 않고 Expression Language 를 사용하여 생성)

 

${now():format('yyyyMMdd')}

 

 

 

 

 

 

 

UnpackContent

- 플로우 내에서 압축파일을 해제할 때 사용한다. 플로우 내에서만 압축해제 된 상태로 남아 있다.

Packaging Format : 압축할 포맷 지정 ( zip , tar)

File Filter : 필터링할 파일 지정  .* (모든 파일)

 

 

 

 

 

 

 

 

 

RouteOnAttribute

- 플로우파일 분리 프로세스

- 플로우 파일 내의 데이터가 여러개인 경우 플로우가 가진 속성을 기준으로 플로우파일을 분리 할 때 사용

Routing Strategy : 분기할 기준 선택

 

filename 이라는 속성을 생성하여, filename에 ${time} 속성(오늘날짜) 포함하는 이름을 가진 파일들만 분리하도록 한다.

${filename:contains(${time})}

 

 

 

 

 

 

 

 

CalculateRecordStats

- cvs(정형데이터)를 수집 및 적재하는 경우 데이터 건수를 측정할 수 있는 프로세스

 cvs데이터가 형식에 맞게 되어있는지 검증할 때도 사용할 수 있다.

 

 

 

 

 

 

 

QueryRecord

- 플로우파일 내의 데이터를 쿼리할 때 사용하는 프로세스

정형데이터(cvs), 반정형데이터(json,xml)에서만 사용이 가능하다.

- 사용자지정 속성을 만들어 사용하여 속성의 값으로 데이터베이스에서 조회하는 것과 동일하게 SELECT문을 작성 가능

(대신! FROM절에는 테이블명이 아닌 flowfile을 적어준다. 플로우파일에서 조회하는것이므로..!)

 

 

select
"상가업소번호" AS house_no,
"상호명" AS cmpny_nm,
"지점명" AS point_nm,
"상권업종대분류코드" AS industry_lgcls_cd,
"상권업종대분류명" AS industry_lgcls_nm,
"상권업종중분류코드" AS industry_mdcls_cd,
"상권업종중분류명" AS industry_mdcls_nm,
"상권업종소분류코드" AS industry_smcls_cd,
"상권업종소분류명" AS industry_smcls_nm,
"표준산업분류코드" AS std_cls_cd,
"표준산업분류명" AS std_cls_nm,
"시도코드" AS sd_cd,
"시도명" AS sd_nm,
"시군구코드" AS sgg_cd,
"시군구명" AS sgg_nm,
"행정동코드" AS hd_cd,
"행정동명" AS hd_nm,
"법정동코드" AS bd_cd,
"법정동명" AS bd_nm,
"지번코드" AS addr_cd,
"대지구분코드" AS site_cd,
"대지구분명" AS site_se_nm,
"지번본번지" AS addr_main_no,
"지번부번지" AS addr_sub_no,
"지번주소" AS addr_data,
"도로명코드" AS road_cd,
"도로명" AS road_nm,
"건물본번지" AS bud_main_no,
"건물부번지" AS bud_sub_no,
"건물관리번호" AS bud_mng_no,
"건물명" AS bud_nm,
"도로명주소" AS bud_addr,
"구우편번호" AS old_zip_cd,
"신우편번호" AS new_zip_cd,
"동정보" AS dong_info,
"층정보" AS flor_info,
"호정보" AS room_info,
replace("경도", '.',' ') AS logtd, 
replace("위도", '.',' ') AS lattd,
'${record.count}' AS data_co,
'${time}' AS load_de
from flowfile

파일의 컬럼명을 위와 같이 변경해주었다.

컬럼 중 경도, 위도 컬럼의 값에서 '.' 를 ' '(공백)으로 대체하였다.

측정한 데이터 건수의 값이 들어가는 data_co 컬럼 생성

 time 속성의 값이 들어가는 load_de 컬럼 생성

 

 

 

 

 

 

 

 

 

 

 

 

PutSQL

- 플로우 내에서 데이터베이스에 INSERT 또는 UPDATE 명령을 실행할 때 사용

 

- Support Fragmented Transactions : 단편화된 트랜잭션 지원 여부

true일 경우 프로세서가 가장 먼저 해당 FlowFile의 fragment.identifier 및 fragment.count 특성을 확인하여

fragment.count 값이 1보다 크면 fragment.identifier를 가진 모든 FlowFiles를 단일 트랜잭션으로 처리하고

false로 설정하면 여러 개의 플로우파일에 대한 업데이트 실행 시 독립적으로 수행됨

 

SQL Stataement : 실행할 쿼리문을 작성한다. Delete도 실행 가능!

 

Delete from nifi.br_business_dist_data

 

postgresql DB nifi 스키마 내의 br_business_dist_data 테이블의 데이터를 DELETE 

 

 

 

 

 

 

PutDatabaseRecord

데이터베이스에 플로우파일의 데이터를 적재할 때 사용하는 프로세스

 

postgresql DB nifi 스키마 내의 br_business_dist_data 테이블에 수집한 데이터 INSERT

 

 

 

 

 

 

 

 

 

MergeContent

-플로우에 들어오는 데이터파일들을 압축할 때 사용하는 프로세스. 파일 단위의 결합

 

Merge Strategy : 압축방식

Merge Format : 압축 포맷

Attribute Strategy : 속성 유지 방식

Minimum Number of Entries : 레코드 최소값

Maximum Number of Entries : 레코드 최대값

Minimum Group Size : 결합 최소 사이즈

Max Bin Age : 최대 경과 시간

 

nifi 스키마 내의 br_business_dist_data 테이블에 적재 완료된 데이터들을 zip 포맷으로 압축

 

 

 

 

 

 

UpdateAttribute

- 플로우파일이 가지고 있는 속성의 값을 변경 또는 새로운 속성을 플로우파일에 등록할 수 있는 프로세스

  주로 파일명을 변경하거나 파일적재시 필요한 변수를 생성할 때 사용한다.

 

Store State : 상태를 저장할 지 여부를 선택

Cache Value Lookup Cache Size : 캐시에 저장할 표준 조회 값 수를 지정

 

압축된 파일의 파일명을 br_오늘날짜.zip 로 변경

br_${time}.zip

 

 

 

 

 

 

 

 

 

 

 

PutSFTP

- 수집한 데이터파일을 다른 서버 내 특정 디렉토리에 SFTP로 접근하여 적재하는 경우 사용하는 프로세스

- 해당 프로세스는 같은 서버 내에서 작업하는 경우만 요효하며 다른 서버 내 디렉토리에 적재는 불가하다.

 

Hostname : 서버의 호스트명

Port : 서버 포트번호

Username : SFTP 서버의  계정명

Remote Path : 서버의 적재할 디렉토리명(경로)

Create Directory : 서버에 작성한 디렉토리가 없는 경우 생성할지 여부 true/false

Confict Resolution :  적재시 중복된 파일이 있는 경우 처리할 방법 선택

 

파일을 SFTP서버의 /data01/result_data/ 디렉토리 하위에 이니셜로 폴더를 만들어 적재

 

 

 

 

 

 

 

SFTP 서버의 설정한 디렉토리가 생성되고 파일이 서버에 적재된 모습이다.

 

 

 

 

 

 

 

 

 

후기

 

Linux를 처음으로 활용하여 시나리오를 따라 데이터를 적재 해보았다.

프로그램을 셋팅하는데 시간이 걸렸고, 아직 리눅스 환경에 익숙해지지 못했다.

 

데이터를 수정하는 부분에서 경도와 위도 컬럼의 값에서 .을 공백으로 대체할 때

DB에서 실행해보았을 때는 되길래, 그대로 쿼리를 복사해서 붙여넣기를 했는데 flowfile에서 " "를 붙여서

replace 함수를 실행해야 했었다.

 

 

postgresql DB nifi 스키마 내의 br_business_dist_data 테이블의 데이터를 DELETE를 해줄때, PutSQL 프로세스를 사용해야 했고,

postgresql DB nifi 스키마 내의 br_business_dist_data 테이블에 수집한 데이터 INSERT할 때는 PutdatabaseRecord 프로세스를 사용햐야 했다.

나는 둘을 바꿔서 사용했다... 이유는 PutdatabaseRecord 프로세스에 delete를 할 수 있도록 선택할 수 있는 속성이 있었기 기 때문에 사용했었다. 정말 단순한 이유ㅎ.. 

 

UnpackContent, RouteOnAttribute, MergeContent, PutSFTP 프로세스를 처음 사용해 보았는데, 사용하면서 어떤 프로세스인지 알아보면서 적용하느라 시간이 더 오래 걸렸던거 같다