안녕하십니까!

작업은 다음과 같습니다. 위 그림에 표시된 흐름이 있으며, 이를 N개의 서버로 롤아웃해야 합니다. 아파치 나이파이. 흐름 테스트 - 파일이 생성되어 다른 NiFi 인스턴스로 전송됩니다. 데이터 전송은 NiFi 사이트 간 프로토콜을 사용하여 발생합니다.

NiFi 사이트 간(S2S)은 NiFi 인스턴스 간에 데이터를 전송하는 안전하고 사용자 정의가 가능한 방법입니다. S2S 작동 방식 보기 선적 서류 비치 S2S가 볼 수 있도록 NiFi 인스턴스를 설정하는 것을 기억하는 것이 중요합니다. 여기에.

S2S를 사용한 데이터 전송의 경우 한 인스턴스를 클라이언트라고 하고 두 번째 인스턴스를 서버라고 합니다. 클라이언트는 데이터를 보내고, 서버는 이를 받습니다. 그들 사이의 데이터 전송을 설정하는 두 가지 방법:

1. 푸시. 데이터는 RPG(원격 프로세스 그룹)를 사용하여 클라이언트 인스턴스에서 전송됩니다. 서버 인스턴스에서는 입력 포트를 사용하여 데이터를 수신합니다.
2. 당기세요. 서버는 RPG를 사용하여 데이터를 수신하고, 클라이언트는 출력 포트를 사용하여 전송합니다.

롤링 흐름은 Apache 레지스트리에 저장됩니다.

Apache NiFi 레지스트리는 흐름 저장소 및 버전 관리 도구를 제공하는 Apache NiFi의 하위 프로젝트입니다. 일종의 GIT. 레지스트리 설치, 구성 및 작업에 대한 정보는 다음에서 찾을 수 있습니다. 공식 문서. 저장을 위한 흐름은 프로세스 그룹으로 결합되어 이러한 형태로 레지스트리에 저장됩니다. 이 기사의 뒷부분에서 이에 대해 다시 설명하겠습니다.

처음에는 N이 작은 숫자일 때 합리적인 시간에 흐름을 수동으로 전달하고 업데이트합니다.

그러나 N이 성장함에 따라 더 많은 문제가 발생합니다.

1. 흐름을 업데이트하는 데 더 많은 시간이 걸립니다. 모든 서버로 이동해야 합니다.
2. 템플릿을 업데이트하는 중에 오류가 발생했습니다. 여기에서는 업데이트했지만 여기서는 잊어버렸습니다.
3. 유사한 작업을 많이 수행할 때 인간의 실수

이 모든 것은 프로세스를 자동화하는 것이 필요하다는 사실을 알려줍니다. 이 문제를 해결하기 위해 다음과 같은 방법을 시도했습니다.

1. NiFi 대신 MiNiFi 사용
2. 니파이 CLI
3. NiPyAPI

MiNiFi 사용

ApacheMiNify Apache NiFi의 하위 프로젝트입니다. MiNiFy는 NiFi와 동일한 프로세서를 사용하는 컴팩트 에이전트로, NiFi와 동일한 흐름을 생성할 수 있습니다. MiNiFy에는 흐름 구성을 위한 그래픽 인터페이스가 없기 때문에 에이전트가 가벼워졌습니다. MiNiFy에 그래픽 인터페이스가 없다는 것은 minifi에서 흐름 전달 문제를 해결해야 한다는 것을 의미합니다. MiNiFy는 IOT에서 활발하게 사용되기 때문에 구성 요소가 많고 최종 미니파이 인스턴스에 흐름을 전달하는 프로세스가 자동화되어야 합니다. 익숙한 작업이죠?

또 다른 하위 프로젝트인 MiNiFi C2 Server가 이 문제를 해결하는 데 도움이 될 것입니다. 이 제품은 배포 아키텍처의 중심점이 되도록 만들어졌습니다. 환경 구성 방법 - 설명: 이 기사 Habré에 대한 정보는 문제를 해결하기에 충분합니다. MiNiFi는 C2 서버와 함께 구성을 자동으로 업데이트합니다. 이 접근 방식의 유일한 단점은 C2 서버에서 템플릿을 생성해야 한다는 것입니다. 레지스트리에 대한 간단한 커밋만으로는 충분하지 않습니다.

위 기사에 설명된 옵션은 작동하고 구현하기 어렵지 않지만 다음 사항을 잊어서는 안됩니다.

1. minifi에는 nifi의 모든 프로세서가 포함되어 있지 않습니다.
2. Minifi의 CPU 버전은 NiFi의 CPU 버전보다 뒤떨어집니다.

이 글을 쓰는 시점에서 NiFi의 최신 버전은 1.9.2입니다. 최신 MiNiFi 버전의 프로세서 버전은 1.7.0입니다. MiNiFi에 프로세서를 추가할 수 있지만 NiFi와 MiNiFi 프로세서 간의 버전 불일치로 인해 작동하지 않을 수 있습니다.

니파이 CLI

에 의해 심사 설명 공식 웹사이트에 있는 도구는 흐름 전달 또는 프로세스 관리 분야에서 NiFI와 NiFi 레지스트리 간의 상호 작용을 자동화하는 도구입니다. 시작하려면 이 도구를 다운로드하세요. 따라서.

유틸리티 실행

./bin/cli.sh
           _     ___  _
 Apache   (_)  .' ..](_)   ,
 _ .--.   __  _| |_  __    )
[ `.-. | [  |'-| |-'[  |  /  
|  | | |  | |  | |   | | '    '
[___||__][___][___] [___]',  ,'
                           `'
          CLI v1.9.2

Type 'help' to see a list of available commands, use tab to auto-complete.

레지스트리에서 필요한 흐름을 로드하려면 바스켓의 식별자(버킷 식별자)와 흐름 자체(흐름 식별자)를 알아야 합니다. 이 데이터는 cli 또는 NiFi 레지스트리 웹 인터페이스를 통해 얻을 수 있습니다. 웹 인터페이스는 다음과 같습니다.

CLI를 사용하면 다음을 수행할 수 있습니다.

#> registry list-buckets -u http://nifi-registry:18080

#   Name             Id                                     Description
-   --------------   ------------------------------------   -----------
1   test_bucket   709d387a-9ce9-4535-8546-3621efe38e96   (empty)

#> registry list-flows -b 709d387a-9ce9-4535-8546-3621efe38e96 -u http://nifi-registry:18080

#   Name           Id                                     Description
-   ------------   ------------------------------------   -----------
1   test_flow   d27af00a-5b47-4910-89cd-9c664cd91e85

레지스트리에서 가져오기 프로세스 그룹을 실행합니다.

#> nifi pg-import -b 709d387a-9ce9-4535-8546-3621efe38e96 -f d27af00a-5b47-4910-89cd-9c664cd91e85 -fv 1 -u http://nifi:8080

7f522a13-016e-1000-e504-d5b15587f2f3

중요한 점은 모든 nifi 인스턴스를 프로세스 그룹을 롤백하는 호스트로 지정할 수 있다는 것입니다.

중지된 프로세서와 함께 추가된 프로세스 그룹, 시작해야 함

#> nifi pg-start -pgid 7f522a13-016e-1000-e504-d5b15587f2f3 -u http://nifi:8080

좋습니다. 프로세서가 시작되었습니다. 하지만 문제의 조건에 따라 다른 인스턴스로 데이터를 전송하려면 NiFi 인스턴스가 필요합니다. 데이터를 서버로 전송하기 위해 Push 메서드를 선택했다고 가정해 보겠습니다. 데이터 전송을 구성하려면 이미 흐름에 포함되어 있는 추가된 원격 프로세스 그룹(RPG)에서 데이터 전송을 활성화(전송 활성화)해야 합니다.

CLI 및 기타 소스의 문서에서 데이터 전송을 활성화하는 방법을 찾지 못했습니다. 어떻게 하는지 아시는 분은 댓글에 적어주세요.

bash가 있고 끝까지 갈 준비가 되었으므로 탈출구를 찾을 것입니다! NiFi API를 사용하여 이 문제를 해결할 수 있습니다. 다음 방법을 사용하여 위의 예에서 ID를 가져옵니다(이 경우 7f522a13-016e-1000-e504-d5b15587f2f3). NiFi API 방법에 대한 설명 여기에.

본문에서는 다음 형식의 JSON을 전달해야 합니다.

{
    "revision": {
	    "clientId": "value",
	    "version": 0,
	    "lastModifier": "value"
	},
    "state": "value",
    "disconnectedNodeAcknowledged": true
}

"작동"을 위해 채워야 하는 매개변수:
상태 — 데이터 전송 상태. 데이터 전송을 활성화하려면 TRANSMITTING을 사용하고, 비활성화하려면 STOPPED를 사용하세요.
버전 - 프로세서 버전

버전은 생성 시 기본적으로 0으로 설정되지만 이러한 매개변수는 다음 메서드를 사용하여 얻을 수 있습니다.

Bash 스크립트를 좋아하는 사람들에게는 이 방법이 적합해 보일 수 있지만 나에게는 어렵습니다. Bash 스크립트는 제가 가장 좋아하는 스크립트가 아닙니다. 제 생각에는 다음 방법이 더 흥미롭고 편리합니다.

NiPyAPI

NiPyAPI는 NiFi 인스턴스와 상호작용하기 위한 Python 라이브러리입니다. 문서 페이지 라이브러리 작업에 필요한 정보가 포함되어 있습니다. 빠른 시작은 다음에 설명되어 있습니다. 프로젝트 github에서.

구성 롤아웃을 위한 스크립트는 Python 프로그램입니다. 코딩으로 넘어 갑시다.
추가 작업을 위해 구성을 설정합니다. 다음 매개변수가 필요합니다.

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #путь до nifi-api инстанса, на котором разворачиваем process group
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #путь до nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' #название registry, как будет называться в инстансе nifi
nipyapi.config.bucket_name = 'BucketName' #название bucket, из которого подтягиваем flow
nipyapi.config.flow_name = 'FlowName' #название flow, которое подтягиваем

또한 설명된 이 라이브러리의 메소드 이름을 삽입하겠습니다. 여기에.

다음을 사용하여 레지스트리를 nifi 인스턴스에 연결합니다.

nipyapi.versioning.create_registry_client

이 단계에서는 레지스트리가 이미 인스턴스에 추가되었는지 확인하는 방법을 추가할 수도 있습니다. 이를 위해 다음 방법을 사용할 수 있습니다.

nipyapi.versioning.list_registry_clients

바구니의 흐름을 추가로 검색하기 위해 버킷을 찾습니다.

nipyapi.versioning.get_registry_bucket

발견된 버킷에 따라 흐름을 찾고 있습니다.

nipyapi.versioning.get_flow_in_bucket

다음으로, 이 프로세스 그룹이 이미 추가되었는지 이해하는 것이 중요합니다. 프로세스 그룹은 좌표에 따라 배치되며 두 번째 프로세스 그룹이 프로세스 그룹 위에 겹쳐지는 상황이 발생할 수 있습니다. 확인해보니 그럴 수 있습니다 🙂 추가된 모든 프로세스 그룹을 얻으려면 다음 메소드를 사용하세요.

nipyapi.canvas.list_all_process_groups

예를 들어 이름으로 검색할 수 있습니다.

템플릿을 업데이트하는 과정은 설명하지 않고 새 버전의 템플릿에 프로세서가 추가되면 대기열에 메시지가 있는 데 문제가 없다고만 말씀드리겠습니다. 그러나 프로세서가 제거되면 문제가 발생할 수 있습니다(nifi는 메시지 큐가 앞에 쌓인 경우 프로세서 제거를 허용하지 않습니다). 제가 이 문제를 어떻게 해결했는지에 관심이 있으시면 제게 편지를 보내 주시면 이 점에 대해 논의하겠습니다. 기사 끝 부분에 연락처가 있습니다. 프로세스 그룹을 추가하는 단계로 넘어가겠습니다.

스크립트를 디버깅할 때 최신 버전의 흐름이 항상 풀업되지 않는 기능을 발견했으므로 먼저 이 버전을 명확히 하는 것이 좋습니다.

nipyapi.versioning.get_latest_flow_ver

배포 프로세스 그룹:

nipyapi.versioning.deploy_flow_version

프로세서를 시작합니다.

nipyapi.canvas.schedule_process_group

CLI에 대한 블록에서 원격 프로세스 그룹에서는 데이터 전송이 자동으로 활성화되지 않는다고 기록되어 있습니까? 스크립트를 구현할 때에도 이 문제가 발생했습니다. 그 당시에는 API를 사용하여 데이터 전송을 시작할 수 없었고 NiPyAPI 라이브러리 개발자에게 편지를 보내 조언/도움을 요청하기로 결정했습니다. 개발자가 제게 대답했고, 우리는 문제에 대해 논의했고 그는 "뭔가를 확인할" 시간이 필요하다고 썼습니다. 그리고 이제 며칠 후 내 시작 문제를 해결하는 Python 함수가 작성된 이메일이 도착했습니다 !!! 당시 NiPyAPI 버전은 0.13.3이었고 물론 그 버전에는 아무것도 없었습니다. 하지만 최근 출시된 0.14.0 버전에서는 이 기능이 이미 라이브러리에 포함되어 있습니다. 만나다

nipyapi.canvas.set_remote_process_group_transmission

따라서 NiPyAPI 라이브러리의 도움으로 레지스트리를 연결하고 흐름을 롤업했으며 프로세서와 데이터 전송도 시작했습니다. 그런 다음 코드를 정리하고, 모든 종류의 검사를 추가하고, 로깅하는 것이 전부입니다. 그러나 그것은 완전히 다른 이야기입니다.

내가 고려한 자동화 옵션 중에서 후자가 가장 효율적인 것 같았습니다. 첫째, 이것은 보조 프로그램 코드를 포함하고 프로그래밍 언어의 모든 이점을 누릴 수 있는 Python 코드입니다. 둘째, NiPyAPI 프로젝트가 활발히 개발 중이므로 문제가 있는 경우 개발자에게 문의할 수 있습니다. 셋째, NiPyAPI는 복잡한 문제를 해결하기 위해 NiFi와 상호 작용하는 보다 유연한 도구입니다. 예를 들어 메시지 대기열이 이제 흐름에서 비어 있는지 여부와 프로세스 그룹을 업데이트할 수 있는지 여부를 결정합니다.

그게 다야. 나는 개발자가 직면할 수 있는 함정인 NiFi에서 흐름 전달을 자동화하는 3가지 접근 방식을 설명하고 전달 자동화를 위한 작동 코드를 제공했습니다. 당신도 나만큼 이 주제에 관심이 있다면 - 감사합니다!

출처 : habr.com