두 개의 CSV를 Apache Nifi와 결합하는 방법

Question

저는 Talend와 같은 ETL 도구를 찾고 Apache Nifi를 사용할 수 있는지 조사하고 있습니다. Nifi는 다음을 수행하는 데 사용할 수 있습니다 :

공통 컬럼에 CSV를 참여

디스크에 결합 된 CSV를 작성

1. 는 로컬 디스크에 배치 된 두 개의 CSV 파일을 선택

Nifi에서 작업 설정을 시도했지만 두 개의 개별 CSV 파일을 결합하는 방법을 볼 수 없었습니다. 이 작업은 Apache Nifi에서 가능합니까?

QueryDNS processor은 다른 CSV 파일을 사용하여 하나의 CSV 파일을 풍부하게 만드는 데 사용할 수 있지만이 사용 사례에서는 지나치게 복잡해 보입니다. 여기

는 STATE_ID에 결합 될 필요가 입력 CSV 형식의 예이다 :

입력 파일

customers.csv

id | name | address  | state_id 
---|------|--------------|--------- 
1 | John | 10 Blue Lane | 100 
2 | Bob | 15 Green St. | 200 

states.csv을

state_id | state 
---------|--------- 
100  | Alabama 
200  | New York 

출력 파일

output.csv

id | name | address  | state 
---|------|--------------|--------- 
1 | John | 10 Blue Lane | Alabama 
2 | Bob | 15 Green St. | New York 

Answer 1

아파치 NiFi는 더 많은 데이터 흐름 도구이며 실제로 스트리밍 데이터의 임의의 조인을 수행하지 않습니다. 일반적으로 이러한 유형의 작업은 Storm, Flink, Apex 등과 같은 스트림 처리 시스템 또는 ETL 도구에 더 적합합니다.

NiFi가 잘 수행 할 수있는 조인 유형은 고정 크기 조회 데이터 집합이있는 경우의 보완 검색과 수신 데이터의 각 레코드에 대해 조회 값 세트를 사용하여 값을 검색하는 경우입니다. 예를 들어, 모든 상태를 포함하는 파일을 가리키는 "State Data"속성을 가진 LookUpState라는 프로세서가있을 수 있습니다. 그러면 customers.csv가이 프로세서의 입력이 될 수 있습니다. https://github.com/jfrazee/nifi-lookup-service

:

커뮤니티 회원은 NiFi에 대한 일반적인 조회 서비스를 만들기 위해 프로젝트를 시작

Answer 2

이 NiFi에서 맵 캐시 제어기 서비스에 설정된 기준을로드하는 것 하나는 다음과 일반적인 패턴. 이 경우에는 states.csv 데이터입니다. 그런 다음 고객 데이터의 라이브 피드가 들어오고 ReplaceText과 같은 것을 사용하여이 참조 데이터가 풍부 해 지거나 Groovy에 사용자 정의 프로세서를 작성할 수도 있습니다. 이것을 조각 낼 수있는 방법이 많이 있습니다. 심지어 더 쉽게 만들기 위해 오는 JIRA/PR도 있습니다. Apache Storm, Spark 및 Flink와 같은 처리 시스템에서 가장 잘 수행되는 라이브 스트림 조인 요소가 있지만, 언급 한 경우 NiFi에서 잘 수행 할 수 있습니다.