Apache NiFi

1. Apache NiFi

- Apache NiFi Introduction

서로 다른 시스템 간에 데이터를 처리하고 배포하는 강력하고 사용하기 쉬우며 안정적인 시스템이며 NSA에서 Apache에 기증한 Dataflow엔진
데이터를 가져오고 이를 처리 후 적재하기 위한 ETL Tools의 일종으로 분산환경에서 대량의 데이터를 수집,처리하며 FBP개념을 구현하여 만든 오픈소스
서로 다른 소스와 대상 시스템 간의 데이터를 전송하고 관리할 수 있는 실시간 데이터 수집 플랫폼

- Apache NiFi General Features

Apache Nifi는 시스템 간 데이터 전달을 효율적으로 처리, 관리, 모니터링 하기에 최적화 된 시스템, 대량의 데이터 수집, 처리가능
Dataflow를 쉽게 구성이 가능하며, 시스템 간의 데이터 이동과 내용을 볼 수 있는 UI제공
강력한 자원과 권한 관리를 통해 Multi-tenant(단일로 여러 소프트웨어에 공급하는 것)지원
데이터 어느 시스템으로 왔는지 추적 가능
NiFi 시스템 간 통신 지원(site to site)
SSL, HTTPS, SSH 및 암호화 및 SFTP, HDFS, KAFKA 등 프로토콜 지원
사용자 및 역할 관리를 지원하며 인증을 위해 LDAP 구성 가능

- Apache NiFi 장점

SFTP를 사용하여 원격 시스템에서 데이터를 가져오고 데이터 계보를 보장
클러스터링을 지원하므로 동일한 흐름으로 다른 데이터를 처리하는 여러 노드에서 작동할 수 있어 데이터 처리 성능이 향상
사용자 수준, 프로세스 그룹 수준 및 기타 모듈에 대한 보안 정책 제공
UI는 HTTPS에서도 실행할 수 있으므로 NiFi와 사용자의 상호 작용을 안전하게 보호
약 188개의 프로세서를 지원하며 사용자는 다양한 데이터 시스템을 지원하는 맞춤형 플러그인 생성 가능

- Apache NiFi 단점

사용자가 노드를 변경하는 동안 NiFi 클러스터에서 노드 연결이 끊어지면 flow.xml이 무효화되어 관리자가 연결된 노드에서 flow.xml을 수동으로 복사하지 않는 한 Anode는 클러스터에 다시 연결 불가
기본 노드 전환의 경우 상태 지속성 문제가 있어 프로세서가 소싱 시스템에서 데이터를 가져올 수 없는 경우가 발생

- Apache NiFi 주요 개념

(FlowFile, FlowFile Processor, Connection, Flow Controller, Process Group)

- Apache NiFi 아키텍처

Web server
: UI 웹 서버를 통해 제공, HTTP 기반 요청 및 API 관리
사용자는 이를 이용해 DataFlow 개발, 제어, 모니터링 가능

Flow Controller
: Processor가 어느 간격 또는 시점에 실행 되는지 스케쥴링
스레딩을 통해 다수의 동시작업 지원

Extension
: 개발자가 NiFi 제공 기본 Process 이외 Process 개발해 확장 가능

FlowFile Repository
: Write-Ahead-Log로 FlowFile의 속성과 상태값을 저장
시스템 장애시 데이터가 유실 되지 않도록 주의

Content Repository
: FlowFile의 데이터(Content)가 저장
여러 디렉토리에 분석 저장이 가능 대용량 데이터 처리에 용의

Provenance Repository
: 데이터의 처리 단계별로 FlowFile 데이터를 보관
여러 디스크를 지원하며 각 데이터는 인덱스 검색 가능

• Apache NiFi 클러스터링

• NiFi는 Zero-Master 클러스터링을 채택하여 전체 클러스터가 동일한 작업 수행, 노드들에서 다루어지는 데이터는 Apache Zookeper를 통해 분산처리
  

• Cluster Coordinator
  각 NiFi 서버들의 정보 관리, 새로 추가된 노드에 최근 플로우 제공
  DataFlow의 추가, 수정, 삭제 등의 변경을 클러스터에 등록된 NiFi 노드들에 복제

• Primary Node
  모든 클러스터는 하나의 Primary Node를 가지며, ZooKeeper Server로 부터 선출
  특정 단일 노드에서만 실행하고자 하는 고립된 프로세서를 실행하는 대표 노드
  Primary Node에서 실행된 데이터를 각 노드에 공급해주는 것이 일반적

• ZooKeeper Server
  Primary Node, Cluster Coordinator는 ZooKeeper Server에서 자동으로 선출
  클러스터 내에 NiFi 노드들 중 한대가 자동으로 Primary Node와 Cluster Coordinator가 된다. 이를 Zero-Master Cluster라 한다.

- Apache NiFi 압축 파일 폴더 구조

• 서버 구동 전
  nifi
  ├── bin - batch파일 이나 구동하는 shell파일 폴더
  ├── conf - 기본정보가 있는 폴더 nifi.properties등
  ├── docs - web ui html 및 doc관련 내용 폴더
  ├── extensions - 확장프로그램 저장소
  ├── lib - nifi관련 nar, jar파일 폴더
  ├── LICENSE
  ├── NOTICE
  └── README

• 서버 구동 후 파일 구조 repository폴더들과 log 폴더(generated user id, password 저장된 폴더)가 추가 생성됨

nifi
├── bin - batch파일 이나 구동하는 shell파일 폴더
├── conf - 기본정보가 있는 폴더 nifi.properties등
├── content_repository - flowfile에서 사용하는 content 저장소
├── database_repository
├── docs - web ui html 및 doc관련 내용 폴더
├── extensions - 확장프로그램 저장소
├── flowfile_repository - nifi에서 생성한 dataflow 저장소
├── lib - nifi관련 nar, jar파일 폴더
├── logs - 서버 구동 시 생성되는 user와 app로그 등이 여기 있음
├── provenance_repository - dataflow에 사용하는 content중 검증된 내용 저장/사용
├── run - 서버 구동 정보
├── state
├── work
│ ├── docs - nifi processor에서 사용하는 components들 있음
│ ├── jetty
│ ├── nar - 위에 docs에서 사용하는 compoenets들 nar파일 있음
├── LICENSE
├── NOTICE
└── README

2. Install NiFi on Ubuntu in AWS

- 설치가능한 리스트 및 설치된 패키지 업데이트

$ sudo apt update
$ sudo apt upgrade -y

- Install java

$ sudo apt install openjdk-8-jre-headless -y
$ update-alternatives --config java

- JAVA_HOME 경로 설정

$ sudo vim /etc/profile
. . .
export JAVA_HOME="/usr/lib/jvm/java-8-openjdk-amd64/jre"

export PATH=JAVA_HOME/bin:$PATH . . .$ source /etc/profile

- Install NiFi

$ wget https://archive.apache.org/dist/nifi/1.1.1/nifi-1.1.1-bin.tar.gz
$ tar -xvf nifi-1.1.1-bin.tar.gz
$ cd nifi-1.1.1

- NiFi web server 데몬 실행

$ bin/nifi.sh status # 서버 상태 확인
$ bin/nifi.sh start # 서버 구동

http://인스턴스 퍼블릭 IPv4 주소:8080/nifi 접속

$ bin/nifi.sh stop # 서버 중지

* Apache NiFi Test

• NiFi에서 이벤트를 받아 Flow File을 생성하는 모습 확인
  (Tailing 프로세스를 만들어서 실제 Tailing 된 Flow File 확인)

1. Apache NiFi 접속

http://인스턴스 퍼블릭 IPv4 주소:8080/nifi

2.Processor 추가하기

특정파일 Tailing을 위해 TagFile 프로세스 추가

• tail 프로세서와 log 프로세서 각각 생성
  
  

3. Tail File 설정하기

프로세서 생성 직후 상태 ⚠

• TailFile 세부설정
  : Configure -> PROPERTIES -> Tailing 할 파일 경로 추가
  (ex: /home/ubuntu/nifi-1.1.1/nifi/tailing/a)
  

4. Relationship 설정하기

Tail File에 성공한 Flow File들은 success 큐에 쌓이게 되고 이를 Log Attribute 프로세서가 소비하게 되면 success 큐에서 Flow File들이 빠진다.

5. Log Attribute 설정하기

: Configure -> SETTINGS -> Automatically Terminate Relationships : success
(성공한 Flow File은 자동으로 삭제하도록 설정)

6. 프로세서 실행

7. Tail 이벤트 발생시키기

Tail 이벤트를 발생시키기 위해 파일을 생성하고 텍스트를 입력해본다.
프로세서에서 바라보는 /home/ubuntu/nifi-1.1.1/nifi/tailing 위치로 이동한다.

아래와 같이 의도적으로 파일의 내용은 append 한다.

$ echo "hello world" >> a

8. 이벤트 결과 확인하기

: List queue-> Info 아이콘 -> view

List Queue 이후 아래와 같이 Flow File에 대한 정보 확인 가능

3. Install NiFi in Docker

- 설치가능한 리스트 및 설치된 패키지 업데이트

$ sudo apt update
$ sudo apt upgrade -y

- Install Docker

$ sudo apt install docker.io
$ sudo usermod -a -G docker $USER$ newgrp docker

- Docker image pull

$ docker pull apache/nifi:latest

- Docker Run

Apache NiFi를 도커 컨테이너로 실행한다.
$ docker run --name nifi -p 8443:8443 -e NIFI_WEB_HTTP_PORT='8443' -d apache/nifi:latest

-e 옵션을 사용하여 http접근 포트를 8443으로 지정해주고 host는 따로 설정하지 않음.

아래 명령을 실행하여 Docker 컨테이너에서 NiFi log를 가져온다.
$ docker logs -f nifi

Apache NiFi가 실행중인 포트를 가져오기 위해 다음 행을 찾는다.

• NiFi 프런트엔드가 포트 번호 8443에서 실행 중임을 나타낸다.

임시 사용자 이름과 암호를 확인한다.

• nifi.properties 파일을 확인해보면 기본구성이 단일 사용자 인증을 활성화하고 있다.
  
  생성된 사용자 이름은 36자로 구성된 임의의 UUID이며, 생성된 비밀번호는 32자로 구성된 임의의 문자열이다.
  
  Username과 Password는 log파일에서도 동일하게 확인할수 있다.

사용자 이름과 암호를 변경하고자 한다면 다음 명령어를 사용하면 된다.
$ ./bin/nifi.sh set-single-user-credentials

NiFi Security Configuration ----------------------------------------

NiFi 1.14.0 버전 이상부터 http가 아닌 https가 기본으로 적용되었다.

Nifi.properties파일의 nifi.remote.input.secure 항목에 따라 http or https가 결정된다.(true : https로 요청 / false : http로 요청)

NiFi는 HTTP 또는 HTTPS에서 실행을 지원하지만 동시 지원은 하지 않는다.
따라서 활성화하고자 하는 속성을 제외하고는 설정 해제해주는 것이 중요하다.

위와 같이 https를 활성하고자 한다면 http의 설정은 빈값으로 설정하는 것이다.

만약 모든 네트워크 인터페이스에서 액세스할 수 있도록 하려면 값을 0.0.0.0 사용해야한다.
-> 해당 설정 내용은 /conf/nifi.properties 파일에서 확인할 수 있다.

—----------------------------------------------------------------------------

- NiFi web server 접속

http://인스턴스 퍼블릭 IPv4 주소:8443/nifi 접속한다.

4. Install Apache NiFi with Helm

Ref:https://artifacthub.io/packages/helm/cetic/nifi/0.1.3

- Prerequisites

Kubernetes cluster
Helm 2.8.0+
Pv provisioner support in the underlying infrastructure

- Install Helm

$ curl -fsSL -o get_helm.sh https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3
$ chmod 700 get_helm.sh
$ ./get_helm.sh
$ helm version ## helm 설치 완료 후 버전 확인하기

- Add Helm repository

$ helm repo add cetic http://cetic.gitHub.io/helm-charts
$ helm repo update

- Configure the way how to expose nifi service

Ingress : Kubernetes 클러스터에 Ingress 컨트롤러를 설치해야한다.
ClusterIP : 클러스터 내부 IP에 서비스를 expose한다.
NodePort : 고정포트에서 각 노드의 IP에 대한 서비스를 expose한다. (NodeIP:NodePort)
Load Balancer : 클라우드 제공자의 로드벨런서를 이용하여 서비스를 expose한다.

- Configure the way how to persistent data

Disable : 데이터는 포트 종료후 유지되지 않는다.
Persistent Volume Claim(default) : StorageClass 볼륨을 동적으로 프로비저닝 하려면 Kubernetes 클러스터에 기본값이 필요하다. 사용할 영구 볼륨이 이미 있는 경우 다른 StorageClass를 지정해야한다.

- Repo fetch

추가한 repo에서 필요한 폴더를 가져온다.
$ helm repo list
$ helm search repo cetic
$ helm fetch cetic/nifi
$ tar fvzx nifi-1.1.1.tgz

- pv 생성

$ sudo mkdir /data/volmes/pv1
$ sudo chmod 777 /data/volumes/pv1

$ vim pv1.yaml
apiVersion: v1
kind: PersistentVolume
metadata:
name: pv1
labels:
pv: pv1
spec:
capacity:
storage: 10G
accessModes:

• ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: ""
  local:
  path: /data/volumes/pv1
  nodeAffinity:
  required:
  nodeSelectorTerms:
  - matchExpressions:
  - key: kubernetes.io/hostname
  operator: In
  values:
  - master

$ vim pv2.yaml
apiVersion: v1
kind: PersistentVolume
metadata:
name: pv2
labels:
pv: pv2
spec:
capacity:
storage: 10G
accessModes:

• ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: ""
  local:
  path: /data/volumes/pv2
  nodeAffinity:
  required:
  nodeSelectorTerms:
  - matchExpressions:
  - key: kubernetes.io/hostname
  operator: In
  values:
  - worker01

$ vim pv3.yaml
apiVersion: v1
kind: PersistentVolume
metadata:
name: pv3
labels:
pv: pv3
spec:
capacity:
storage: 10G
accessModes:

• ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: ""
  local:
  path: /data/volumes/pv3
  nodeAffinity:
  required:
  nodeSelectorTerms:
  - matchExpressions:
  - key: kubernetes.io/hostname
  operator: In
  values:
  - worker02

위와같이 pv1,pv2,pv3에 대한 yaml파일을 작성한뒤 apply하여 pv를 생성한다.

$ kubectl apply -f pv1.yaml
$ kubectl apply -f pv2.yaml
$ kubectl apply -f pv3.yaml
$ kubectl get pv #생성된 pv 확인

- Install the chart

my-nifi 이름으로 nifi helm chart를 설치한다.
$ helm install my-nifi cetic/nifi
$ helm status my-nifi #배포된 chart 상태 확인

모든 pod가 Running이며, pv와 pvc가 Bound되어있는지 확인한다.
$ kubectl get all,pv,pvc

- Kubectl port-forwarding

클러스터가 준비가 끝나면 아래와 같이 kubectl port-forwarding을 통해 NiFi UI에 액세스한다.
$ kubectl port-forward -n default svc/my-nifi 8443:8443

- NiFi web server 접속

http://localhost:8443/nifi 접속한다.

- Username / Password 확인

NiFi web server에 접속하면 Log In 창으로 연결된다.

• nifi.properties 파일을 확인해보면 기본구성이 단일 사용자 인증을 활성화하고 있다.
  

지정되어있는 Username과 Password를 확인하기 위해 login-identity-providers.xml 파일을 확인해보면 values.yaml파일에서 값을 받아오는것을 알수있다.

Values.yaml파일에서 아래와 같이 지정된 값을 확인할 수 있다.

Username과 Password를 수정하고 싶다면 해당 파일에서 값을 수정한 뒤 다시 install하여 접속해주면 된다.

지정된 값으로 Log In을 시도하면 다음과 같이 접속이 가능하다.

- Uninstallation

my-nifi 배포를 제거/삭제하려면 다음과 같은 명령어를 사용한다.
$ helm delete --purge my-nifi

5. Ingress 설정

- Install Using Helm

$ helm repo add ingress-nginx https://kubernetes.github.io/ingress-nginx
$ helm repo update

$ helm install ingress-nginx ingress-nginx/ingress-nginx
NAME: ingress-nginx
LAST DEPLOYED: Fri Apr 16 08:29:48 2021
NAMESPACE: default
STATUS: deployed
REVISION: 1
TEST SUITE: None
NOTES:
The ingress-nginx controller has been installed.
It may take a few minutes for the LoadBalancer IP to be available.
You can watch the status by running 'kubectl --namespace default get services -o wide -w ingress-nginx-controller'

An example Ingress that makes use of the controller:

apiVersion: networking.k8s.io/v1beta1
kind: Ingress
metadata:
annotations:
kubernetes.io/ingress.class: nginx
name: example
namespace: foo
spec:
rules:
- host: www.example.com
http:
paths:
- backend:
serviceName: exampleService
servicePort: 80
path: /

# This section is only required if TLS is to be enabled for the Ingress
tls:
    - hosts:
        - www.example.com
      secretName: example-tls

If TLS is enabled for the Ingress, a Secret containing the certificate and key must also be provided:

apiVersion: v1
kind: Secret
metadata:
name: example-tls
namespace: foo
data:
tls.crt:
tls.key:
type: kubernetes.io/tls

$ kubectl get service
NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE
service/ingress-nginx-controller LoadBalancer 10.96.199.175 80:30007/TCP,443:30182/TCP 3m11s
service/ingress-nginx-controller-admission ClusterIP 10.105.9.10 443/TCP 3m11s
service/kubernetes ClusterIP 10.96.0.1 443/TCP 35m

테스트

$ wget https://subicura.com/k8s/code/guide/ingress/echo-v1.yml
$ vim echo-v1.yml
...
- host: 192-168-137-51.sslip.io
...

$ kubectl apply -f echo-v1.yml
ingress.networking.k8s.io/echo-v1 created
deployment.apps/echo-v1 created
service/echo-v1 created

$ kubectl get all
NAME READY STATUS RESTARTS AGE
pod/echo-v1-68597f9b8d-6246j 1/1 Running 0 7m53s
pod/echo-v1-68597f9b8d-pt6c6 1/1 Running 0 7m53s
pod/echo-v1-68597f9b8d-tttmj 1/1 Running 0 7m53s
...
NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE
service/echo-v1 ClusterIP 10.102.5.241 3000/TCP 7m53s
...
NAME READY UP-TO-DATE AVAILABLE AGE
deployment.apps/echo-v1 3/3 3 3 7m53s
...
NAME DESIRED CURRENT READY AGE
replicaset.apps/echo-v1-68597f9b8d 3 3 3 7m53s

$ kubectl get ing
NAME CLASS HOSTS ADDRESS PORTS AGE
echo-v1 192-168-137-51.sslip.io 80 4s

웹브라우저에서 192-168-137-51.sslip.io 주소로 접속

version: v1
hostname: echo-v1-68597f9b8d-tttmj
headers:
host: '192-168-137-51.sslip.io:30007'
x-request-id: 7c8b777c6019fc2578cc0b927d3a6ac8
x-real-ip: 10.244.0.1
x-forwarded-for: 10.244.0.1
x-forwarded-host: '192-168-137-51.sslip.io:30007'
x-forwarded-port: '80'
x-forwarded-proto: http
x-scheme: http
cache-control: max-age=0
upgrade-insecure-requests: '1'
user-agent: >-
Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like
Gecko) Chrome/89.0.4389.90 Safari/537.36
accept: >-
text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,/;q=0.8,application/signed-exchange;v=b3;q=0.9
accept-encoding: 'gzip, deflate'
accept-language: 'ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7'
cookie: _ga=GA1.2.345706462.1618556786; _gid=GA1.2.570985710.1618556786
query: {}

대시보드 인그레스

Ref: https://serverfault.com/questions/1019919/how-to-properly-configure-access-to-kubernees-dashboard-behind-nginx-ingress

$ cat dashboard-ingress.yml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: kubernetes-dashboard
annotations:
nginx.ingress.kubernetes.io/backend-protocol: "HTTPS"
spec:
rules:
- host: 192-168-137-51.sslip.io
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: kubernetes-dashboard
port:
number: 443

$ kubectl apply -f dashboard-ingress.yml

NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE
service/ingress-nginx-controller LoadBalancer 10.96.199.175 80:30007/TCP,443:30182/TCP 24h

웹브라우저에서 https://192-168-137-51.sslip.io:30182/ 로 접속