OCI Data Integration 활용 - Data Integration Task 생성

지난번 Post에서는 Oracle Cloud의 Data Integration의 간단한 소개와 Data Loader Task 생성을 알아보았습니다.
이번 Post에서는 Data Integration의 Data Flow를 이용한 Data Integration Task 생성에 대해 알아보도록 하겠습니다. 

이번 Data Integration Task 시나리오는 Customer 정보가 저장된 customers.json 파일과 매출 정보인 REVENUE.csv 파일을 이용하여 Flow를 생성하고 이를 이용하여 Integration Task를 생성 합니다. 

 

우선 지난번에 생성한 프로젝트인 DI_Workshop에서 Data Flow를 생성합니다. 

 

아래와 같이 Data Flow Designer가 생성되면 임의의 이름을 지정합니다. (ex : Load Customers and Revenue Data) 

 

좌측의 Operator panel에서 Source Operator를 drag&drop으로 canvas로 가져옵니다. 

 

Source Operator의 Detail에서 대상을 Object Storage 상의 Customer.json 파일을 지정합니다. 

 

두번째 Source Operator를 추가합니다. 두번째 source operator는 object storage의 REVENUE.csv file을 지정합니다. 

 

Source Operator로 구성한 Customers와 Revenue에 대해 각각 Filter를 추가해 보겠습니다.

Source인 Revenue.csv에서 ORDER 상태가 Booked 인 데이터만 추출하고 Customer.json에서 미국 고객만 대상으로 지정하겠습니다. 

Operator Panel에서 filter를 Drag&Drop 하여 canvas로 가져옵니다. Revenue와 Filter_1 operator를 연결해 줍니다. 연결은 Source인 Revenue operator의 우측에 커서를 올려놓으면 연결을 위한 circle이 보여지며 해당 circle을 filter로 drag&drop 하면 연결됩니다. 

 

filter_1 operator의 상세 정보에서 filter condition을 생성합니다. 

 

ODER_STATUS에 아래와 같이 '1-Booked' 필터 조건을 입력하고 생성합니다. 

 

Source operator인 Customer에 대해서도 Filter operator를 추가하여 대상 고객을 미국으로 제한합니다. 

 

이제 두개의 Source 를 Join 하기 위해 Join operation을 drag&drop 합니다. 

 

상세 정보에서 Join Type을 inner로 선택하고 아래와 같이 join condition을 생성합니다. 

 

이제 최종 join 된 데이터를 저장할 Target을 구성합니다. operator panel에서 target operator를 drag&drop 하여 추가하고 join operator와 연결합니다.

 

Target operator의 상세 정보를 설정합니다. Target은 Autonomous Datawarehouse instance의 SCOTT 계정으로 지정되었으며 새로운 CUSTOMER_TARGET이라는 테이블을 생성하여 insert 하게 됩니다. 

 

이제 생성된 Data Flow에 대한 validation을 수행합니다. Data Flow Disigner의 우측 상단의 valdate를 수행하여 검토하고 create and close를 수행합니다. 

 

앞서 생성한 Data Flow를 확인하고 이제 생성된 Data Flow를 이용하여 Task를 생성하게 됩니다. 

 

DR_Workshop Project의 Task로 이동하여 Integration Task를 생성합니다. Integration Task는 'Load Customers Lab'으로 하고 Data Flow에서 앞서 생성한 Data Flow를 선택합니다. Validation이 정상적으로 이루어지면 Integration Task를 생성합니다. 

 

생성된 Integration Task를 Application에 publish 하고 수행합니다. 

 

Job 수행 상태와 데이터를 확인합니다. 

 

이상으로 지난번의 Data Loader Task에 이어 Data Flow를 이용한 Data Integration Task를 생성해 보았습니다. 

다음 Post엔 Publish 된 Task를 이용하여 Pipeline Task를 생성하는 방법에 대해 진행하도록 하겠습니다. 

 

작성자: 조용훈  
개인 시간을 투자하여 작성된 글로서, 글의 내용에 오류가 있을 수 있으며, 글 속의 의견은 개인적인 의견입니다.