Apache NIFI - 실제 기회에 대한 간략한 개요

소개

마침 현재 직장에서 이 기술을 접하게 되었습니다. 약간의 배경 지식부터 시작하겠습니다. 다음 회의에서 우리 팀은 다음과 같은 통합을 구축해야 한다는 말을 들었습니다. 알려진 시스템. 통합이란 이 잘 알려진 시스템이 HTTP를 통해 특정 엔드포인트로 요청을 보내고, 이상하게도 우리가 SOAP 메시지 형식으로 응답을 다시 보내는 것을 의미했습니다. 모든 것이 단순하고 사소해 보입니다. 이것으로부터 당신이 필요로하는 것은 다음과 같습니다 ...

태스크

3개의 서비스를 만듭니다. 그 중 첫 번째는 데이터베이스 업데이트 서비스입니다. 이 서비스는 타사 시스템에서 새로운 데이터가 도착하면 데이터베이스의 데이터를 업데이트하고 CSV 형식의 파일을 생성하여 다음 시스템으로 전송합니다. 두 번째 서비스의 끝점은 전송된 파일을 수신하여 유효성을 검사하고 FTP를 통해 파일 저장소에 저장하는 FTP 전송 서비스라고 합니다. 세 번째 서비스인 소비자 데이터 전송 서비스는 처음 두 서비스와 비동기식으로 작동합니다. 위에서 설명한 파일을 수신하라는 요청을 제1000자 외부 시스템으로부터 수신하고, 준비된 응답 파일을 가져와 수정한 다음(id, 설명, linkToFile 필드 업데이트) SOAP 메시지 형식으로 응답을 보냅니다. 즉, 전체적인 그림은 다음과 같습니다. 처음 두 서비스는 업데이트할 데이터가 도착한 경우에만 작업을 시작합니다. 세 번째 서비스는 정보 소비자가 많아 분당 약 XNUMX건의 데이터 요청이 있기 때문에 지속적으로 작동합니다. 서비스는 지속적으로 제공되며 해당 인스턴스는 테스트, 데모, 사전 프로덕션, 프로덕션 등 다양한 환경에 위치합니다. 다음은 이러한 서비스의 작동 방식을 보여주는 다이어그램입니다. 불필요한 복잡성을 피하기 위해 일부 세부 사항이 단순화되었음을 즉시 명확히 하겠습니다.

기술심화

문제에 대한 해결책을 계획할 때 우리는 먼저 Spring 프레임워크, Nginx 밸런서, Postgres 데이터베이스 및 기타 기술적인 것들을 사용하여 Java로 애플리케이션을 만들기로 결정했습니다. 기술 솔루션을 개발하면서 이 문제를 해결하기 위한 다른 접근 방식을 고려할 수 있게 되었기 때문에 우리의 시선은 특정 분야에서 유행하는 Apache NIFI 기술에 쏠렸습니다. 이 기술을 통해 우리는 이 3가지 서비스를 알아차릴 수 있었다고 바로 말씀드리겠습니다. 이 글에서는 소비자를 위한 파일 전송 서비스와 데이터 전송 서비스 개발에 대해 설명하겠지만, 글이 도움이 된다면 데이터베이스의 데이터를 업데이트하는 서비스에 대해서도 글을 쓰겠습니다.

이게 뭐예요

NIFI는 데이터의 빠른 병렬 로딩 및 처리, 소스 및 변환을 위한 다수의 플러그인, 구성 버전 관리 등을 위한 분산 아키텍처입니다. 좋은 보너스는 사용하기가 매우 쉽다는 것입니다. getFile, sendHttpRequest 등과 같은 간단한 프로세스는 사각형으로 표시될 수 있습니다. 각 사각형은 프로세스를 나타내며, 그 상호 작용은 아래 그림에서 볼 수 있습니다. 프로세스 설정 상호 작용에 대한 보다 자세한 문서가 작성되었습니다. 여기에 , 러시아어를 사용하는 사람들을 위한 - 여기에. 문서에는 NIFI의 압축을 풀고 실행하는 방법과 사각형이라고도 알려진 프로세스를 만드는 방법이 완벽하게 설명되어 있습니다.
기사를 작성하겠다는 아이디어는 오랜 검색과 수신된 정보를 의식적인 것으로 구조화하는 것뿐만 아니라 미래 개발자의 삶을 조금 더 쉽게 만들고자 하는 열망 끝에 탄생했습니다.

예

사각형이 서로 어떻게 상호 작용하는지에 대한 예가 고려됩니다. 일반적인 방식은 매우 간단합니다. HTTP 요청을 받습니다(이론적으로는 요청 본문에 파일이 포함됩니다. NIFI의 기능을 보여주기 위해 이 예에서는 요청이 로컬 파일 저장소에서 파일을 수신하는 프로세스를 시작합니다). ) 그런 다음 FH에서 파일을 수신하는 프로세스와 FTP를 통해 FH로 이동하는 프로세스를 병행하여 요청이 수신되었다는 응답을 다시 보냅니다. 프로세스가 소위 flowFile을 통해 서로 상호 작용한다는 점을 명확히 할 가치가 있습니다. 이는 속성과 콘텐츠를 저장하는 NIFI의 기본 엔터티입니다. 콘텐츠는 스트림 파일로 표시되는 데이터입니다. 즉, 대략적으로 말하면 한 스퀘어에서 파일을 받아 다른 스퀘어로 전송하면 해당 콘텐츠가 귀하의 파일이 됩니다.

보시다시피, 이 그림은 일반적인 프로세스를 보여줍니다. HandleHttpRequest - 요청을 수락하고, InstallText - 응답 본문을 생성하고, HandleHttpResponse - 응답을 보냅니다. FetchFile - 파일 저장소에서 파일을 받아 정사각형 PutSftp로 전송합니다. - 이 파일을 FTP의 지정된 주소에 넣습니다. 이제 이 과정에 대해 자세히 알아보세요.

이 경우 요청은 모든 것의 시작입니다. 구성 매개변수를 살펴보겠습니다.

StandardHttpContextMap을 제외하고 여기에 있는 모든 것은 매우 간단합니다. 이는 요청을 보내고 받을 수 있는 일종의 서비스입니다. 더 자세히 그리고 예제를 통해서도 다음을 볼 수 있습니다. 여기에

다음으로 사각형의 RefreshText 구성 매개변수를 살펴보겠습니다. replacementValue에 주목할 가치가 있습니다. 이는 응답 형식으로 사용자에게 반환됩니다. 설정에서 로깅 수준을 조정할 수 있고, {nifi의 압축을 푼 위치}/nifi-1.9.2/logs 로그를 볼 수 있으며, 실패/성공 매개변수도 있습니다. 이러한 매개변수를 기반으로 프로세스를 전체적으로 조절할 수 있습니다. . 즉, 텍스트 처리가 성공한 경우에는 사용자에게 응답을 보내는 프로세스가 호출되고, 다른 경우에는 실패한 프로세스를 간단히 기록하게 됩니다.

응답이 성공적으로 생성되었을 때의 상태를 제외하고 HandleHttpResponse 속성에는 특별히 흥미로운 내용이 없습니다.

요청과 응답을 정리했습니다. 이제 파일 수신 및 FTP 서버에 배치하는 단계로 넘어갑니다. FetchFile - 설정에 지정된 경로에서 파일을 받아 다음 프로세스로 전달합니다.

그런 다음 PutSftp 사각형은 파일을 파일 저장소에 배치합니다. 아래에서 구성 매개변수를 볼 수 있습니다.

각 사각형이 실행되어야 하는 별도의 프로세스라는 사실에 주목할 가치가 있습니다. 복잡한 사용자 정의가 필요하지 않은 가장 간단한 예를 살펴보았습니다. 다음으로, 그루브에 약간의 글을 쓰는 좀 더 복잡한 과정을 살펴보겠습니다.

더 복잡한 예

SOAP 메시지를 수정하는 과정으로 인해 소비자에게 데이터를 전송하는 서비스가 조금 더 복잡해졌습니다. 일반적인 프로세스는 아래 그림에 나와 있습니다.

여기서 아이디어도 특별히 복잡하지 않습니다. 소비자로부터 데이터가 필요하다는 요청을 받고 메시지를 받았다는 응답을 보내고 응답 파일 수신 프로세스를 시작한 다음 특정 논리로 편집한 다음 SOAP 메시지 형식으로 파일을 소비자에게 서버로 전송했습니다.

위에서 본 사각형을 다시 설명할 필요는 없을 것 같습니다. 바로 새로운 사각형으로 넘어가겠습니다. 파일을 편집해야 하는데 일반 RectText 유형 사각형이 적합하지 않은 경우 스크립트를 직접 작성해야 합니다. 이는 ExecuteGroogyScript 사각형을 사용하여 수행할 수 있습니다. 해당 설정은 아래에 나와 있습니다.

이 사각형에 스크립트를 로드하는 데는 두 가지 옵션이 있습니다. 첫 번째는 스크립트가 포함된 파일을 다운로드하는 것입니다. 두 번째는 scriptBody에 스크립트를 삽입하는 것입니다. 내가 아는 한, ExecuteScript 스퀘어는 여러 언어를 지원합니다. 그 중 하나는 그루비입니다. 나는 Java 개발자를 실망시킬 것입니다. 그러한 사각형에서는 Java로 스크립트를 작성할 수 없습니다. 정말로 원하는 사람들은 자신만의 맞춤형 사각형을 만들어 NIFI 시스템에 추가해야 합니다. 이 전체 작업에는 탬버린과 함께 꽤 긴 춤이 수반되는데, 이 기사에서는 다루지 않겠습니다. 저는 그루비 언어를 선택했습니다. 다음은 SOAP 메시지의 ID를 점진적으로 업데이트하는 테스트 스크립트입니다. 주의하는 것이 중요합니다. flowFile에서 파일을 가져와 업데이트합니다. 파일을 다시 거기에 넣어 업데이트해야 한다는 점을 잊지 마세요. 모든 라이브러리가 포함되지는 않는다는 점도 주목할 가치가 있습니다. libs 중 하나를 가져와야 하는 경우가 발생할 수 있습니다. 또 다른 단점은 이 사각형의 스크립트를 디버깅하기가 매우 어렵다는 것입니다. NIFI JVM에 연결하고 디버깅 프로세스를 시작하는 방법이 있습니다. 개인적으로 저는 로컬 애플리케이션을 시작하고 세션에서 파일 수신을 시뮬레이션했습니다. 로컬에서도 디버깅을 했습니다. 스크립트를 로드할 때 나타나는 오류는 Google에서 매우 쉽게 확인할 수 있으며 NIFI 자체에서 로그에 기록됩니다.

import org.apache.commons.io.IOUtils
import groovy.xml.XmlUtil
import java.nio.charset.*
import groovy.xml.StreamingMarkupBuilder

def flowFile = session.get()
if (!flowFile) return
try {
    flowFile = session.write(flowFile, { inputStream, outputStream ->
        String result = IOUtils.toString(inputStream, "UTF-8");
        def recordIn = new XmlSlurper().parseText(result)
        def element = recordIn.depthFirst().find {
            it.name() == 'id'
        }

        def newId = Integer.parseInt(element.toString()) + 1
        def recordOut = new XmlSlurper().parseText(result)
        recordOut.Body.ClientMessage.RequestMessage.RequestContent.content.MessagePrimaryContent.ResponseBody.id = newId

        def res = new StreamingMarkupBuilder().bind { mkp.yield recordOut }.toString()
        outputStream.write(res.getBytes(StandardCharsets.UTF_8))
} as StreamCallback)
     session.transfer(flowFile, REL_SUCCESS)
}
catch(Exception e) {
    log.error("Error during processing of validate.groovy", e)
    session.transfer(flowFile, REL_FAILURE)
}

사실 광장의 커스터마이징은 여기서 끝이 납니다. 다음으로, 업데이트된 파일은 서버로 파일 전송을 담당하는 스퀘어로 전송됩니다. 다음은 이 사각형에 대한 설정입니다.

SOAP 메시지가 전송되는 방법을 설명합니다. 우리는 어디에 씁니다. 다음으로 이것이 SOAP임을 표시해야 합니다.

호스트 및 작업(soapAction)과 같은 여러 속성을 추가합니다. 저장하고 확인합니다. SOAP 요청을 보내는 방법에 대한 자세한 내용을 볼 수 있습니다. 여기에

NIFI 프로세스를 사용하기 위한 몇 가지 옵션을 살펴보았습니다. 그들은 어떻게 상호 작용하며 실제 이점은 무엇입니까? 고려된 사례는 테스트 사례이며 실제로 전투에서 발생하는 상황과 약간 다릅니다. 이 글이 개발자들에게 조금이나마 도움이 되기를 바랍니다. 관심을 가져주셔서 감사합니다. 질문이 있으면 글을 쓰세요. 나는 대답하려고 노력할 것이다.

출처 : habr.com