이 글은 아래 Data on EKS 솔루션들을 직접 따라해보면서 문서 상에서의 부족한 부분과 배포 시 트러블슈팅해야할 부분들을 채워 작성하는 글 입니다.
참고 : https://awslabs.github.io/data-on-eks/docs/blueprints/streaming-platforms/nifi
What is Apache NiFi?
Apache NiFi는 시스템 간의 데이터 흐름을 자동화하고 관리하기 위한 오픈 소스 소프트웨어입니다. 미국 국가안보국(NSA)에서 개발되었으며, 이후에 Apache 재단에 기부되었습니다.
NiFi의 유즈 케이스는 다음과 같습니다:
- 데이터 수집: 다양한 소스에서 데이터를 수집하여 Hadoop 또는 Datalake와 연동하여 사용
- 실시간 데이터 스트리밍
- 서로 다른 형식 또는 구조 간의 데이터 변환.
- 시스템 통합: 조직 내 서로 다른 시스템을 연결
NiFi는 Flow-Based Programming 개념을 기반으로 합니다. 주요 구성 요소들은 다음과 같습니다.
- FlowFile: 시스템을 통해 이동하는 데이터
- Processor: FlowFile에 대한 실제 작업을 진행
- Connection: 실제 Processor 간의 연결을 하고 Queue의 역할을 수행
- Processor Group: Processor와 Connection들의 집합
NiFi와 가장 많이 비교되는 Airflow와 대조하면 실시간 스트리밍 처리에 유용하고 Airflow가 코드를 기반으로 작업들을 스케줄링한다면 NiFi의 경우 모든 Flow를 UI로 작업해야한다는 차이가 있습니다.
Prerequisites
이 솔루션을 배포하기 위해서는 (1) 도메인 구매, (2) 도메인 구매 후 도메인을 Route53에 등록 및 ACM 인증서 발급을 우선 받아야합니다. 해당 부분에 대한 자세한 내용은 다시 글로 작성해보도록 하겠습니다.
사용 버전
kubectl version
Client Version: v1.31.3
Kustomize Version: v5.4.2
Server Version: v1.27.16-eks-2d5f260
Terraform v1.10.0
helm version
version.BuildInfo{Version:"v3.16.4", GitCommit:"7877b45b63f95635153b29a42c0c2f4273ec45ca", GitTreeState:"dirty", GoVersion:"go1.23.4"}실제로 배포하기
아래와 같이 환경 변수를 셋팅해줍니다.
export TF_VAR_eks_cluster_domain="도메인 이름"
export TF_VAR_acm_certificate_domain="*.도메인 이름"
export TF_VAR_nifi_sub_domain="nifi"
export TF_VAR_nifi_username="admin" TF_VAR_nifi_username 로 설정해준 이름은 이후 UI에 로그인 할 때 사용됩니다.
github에서 그대로 클론해온 테라폼 파일을 이용하여 terraform plan 수행 시 아래와 같이 에러가 발생합니다.
Error: Failed to query available provider packages
│
│ Could not retrieve the list of available versions for provider hashicorp/random: no available releases match the given constraints 3.3.2, >= 3.6.0
│
│ To see which modules are currently depending on hashicorp/random and what versions are specified, run the following command:
│ terraform providersversions.tf 파일의 random의 버전을 수정해줍니다.
## versions.tf
terraform {
required_version = ">= 1.0.0"
required_providers {
aws = {
source = "hashicorp/aws"
version = ">= 3.72"
}
kubernetes = {
source = "hashicorp/kubernetes"
version = ">= 2.10"
}
helm = {
source = "hashicorp/helm"
version = ">= 2.4.1"
}
random = {
source = "hashicorp/random"
version = "3.6.0"
}
}
}terraform apply를 수행하고 난 뒤에 kubeconfig를 업데이트한 뒤 필요한 모든 파드가 running 상태에 있는지 확인합니다.
aws eks --region us-west-2 update-kubeconfig --name nifi-on-eks
user ~/desktop/doeks/data-on-eks/streaming/nifi main daily-2024.11.13 ● kubectl get pods --all-namespaces ✔ 5370 11:19:03
NAMESPACE NAME READY STATUS RESTARTS AGE
amazon-cloudwatch aws-cloudwatch-metrics-7mjsb 1/1 Running 0 15h
amazon-cloudwatch aws-cloudwatch-metrics-9ccnh 1/1 Running 0 15h
amazon-cloudwatch aws-cloudwatch-metrics-hzjkf 1/1 Running 0 15h
aws-for-fluentbit aws-for-fluent-bit-94xzv 1/1 Running 0 15h
aws-for-fluentbit aws-for-fluent-bit-kns9f 1/1 Running 0 15h
aws-for-fluentbit aws-for-fluent-bit-r26s9 1/1 Running 0 15h
cert-manager cert-manager-56fd8fc94d-mzt9l 1/1 Running 0 68m
cert-manager cert-manager-cainjector-7cbfbfb546-hj4t2 1/1 Running 0 68m
cert-manager cert-manager-webhook-7dd5975fb-9xnvz 1/1 Running 0 68m
external-dns external-dns-59dcf776c7-k9jn5 0/1 Running 185 (5m10s ago) 15h
kube-prometheus-stack kube-prometheus-stack-grafana-c6b5549c8-h2zlq 3/3 Running 0 15h
kube-prometheus-stack kube-prometheus-stack-kube-state-metrics-65cfbcb8b9-5hpwj 1/1 Running 0 15h
kube-prometheus-stack kube-prometheus-stack-operator-6fc7779d7b-v854f 1/1 Running 0 15h
kube-prometheus-stack kube-prometheus-stack-prometheus-node-exporter-66hc2 1/1 Running 0 15h
kube-prometheus-stack kube-prometheus-stack-prometheus-node-exporter-8vjc5 1/1 Running 0 15h
kube-prometheus-stack kube-prometheus-stack-prometheus-node-exporter-9rpch 1/1 Running 0 15h
kube-prometheus-stack prometheus-kube-prometheus-stack-prometheus-0 2/2 Running 0 15h
kube-system aws-load-balancer-controller-8548c6cc9c-b8nbj 1/1 Running 0 15h
kube-system aws-load-balancer-controller-8548c6cc9c-kpqpw 1/1 Running 0 15h
kube-system aws-node-77h64 2/2 Running 0 67m
kube-system aws-node-9zcw9 2/2 Running 0 67m
kube-system aws-node-nk2pc 2/2 Running 0 67m
kube-system cluster-autoscaler-aws-cluster-autoscaler-6ddccb7895-vgkd4 1/1 Running 0 15h
kube-system coredns-6489fddbfb-7g7vp 1/1 Running 0 67m
kube-system coredns-6489fddbfb-kgzw9 1/1 Running 0 67m
kube-system ebs-csi-controller-c5ccbb6fd-sdvhl 6/6 Running 0 67m
kube-system ebs-csi-controller-c5ccbb6fd-wvqzl 6/6 Running 0 67m
kube-system ebs-csi-node-cr7gh 3/3 Running 0 67m
kube-system ebs-csi-node-rtq2m 3/3 Running 0 67m
kube-system ebs-csi-node-zks8k 3/3 Running 0 67m
kube-system kube-proxy-29l6f 1/1 Running 0 67m
kube-system kube-proxy-fx4tr 1/1 Running 0 67m
kube-system kube-proxy-kdr54 1/1 Running 0 67m
kube-system metrics-server-6c9b488c8b-x78h8 1/1 Running 0 15h
nifi nifi-0 5/5 Running 0 49m
nifi nifi-1 5/5 Running 0 49m
nifi nifi-2 5/5 Running 0 49m
nifi nifi-registry-0 1/1 Running 0 49m
nifi nifi-zookeeper-0 1/1 Running 0 49m
nifi nifi-zookeeper-1 1/1 Running 0 49m
nifi nifi-zookeeper-2 user ~/desktop/doeks/data-on-eks/streaming/nifi main daily-2024.11.13 ● kubectl get pods -n external-dns
NAME READY STATUS RESTARTS AGE
external-dns-59dcf776c7-lt2x7 0/1 CrashLoopBackOff 7 (61s ago) 13m위와 같이 솔루션을 그대로 배포했음에도 External DNS의 파드가 CrashLoopBackOff 상태에 있는 것을 확인할 수 있습니다. 트러블슈팅을 위해 해당 파드의 로그를 확인해봅니다.
kubectl logs -n external-dns external-dns-59dcf776c7-lt2x7
time="2025-01-07T15:27:51Z" level=fatal msg="records retrieval failed: failed to list hosted zones:
AccessDenied: User: arn:aws:sts::-:assumed-role/nifi_node_group-eks-node-group-20250107150642481800000009/{instance_id}
is not authorized to perform: route53:ListHostedZones
because no identity-based policy allows the route53:ListHostedZones action
status code: 403...nifi_node_group-eks-node-group-20250107150642481800000009 Role에 route53:ListHostedZones 권한이 부족하여 해당 파드가 제대로 실행되지 못하고 있었습니다. 해당 역할에 권한을 추가해주고 나면 다시 해당 파드가 Running 상태로 돌아옵니다.
참고 : https://repost.aws/knowledge-center/eks-set-up-externaldns
예시 작업 실행해보기
해당 솔루션에서 NiFi를 통해 실행하는 예시 task는 Kinesis Stream으로 레코드를 흘려 DynamoDB 테이블로 쓰는 작업입니다.
(1) 비밀번호 가져오기
먼저 NiFi UI에 접속하기 위해서 secret value를 가져옵니다. UI 로그인 비밀번호는 Secretsmanager 서비스에 저장되어있습니다!
aws secretsmanager get-secret-value --secret-id {nifi_secret_id} --region us-west-2 | jq '.SecretString' --raw-output
password
user name : admin(2) 비어있는 Kinesis Stream과 DynamoDB 테이블을 만듭니다.
$ aws kinesis create-stream --stream-name kds-stream-nifi-on-EKS
DynamoDB 테이블을 만들 때는 아래 json 파일을 이용합니다. 솔루션에서 제시하고 있는 JSON 파일이 잘못 작성되어있는 점을 참고해주세요.
### JSONSchemaDynamoDBTABLE.json
{
"TableName": "NifiStreamingTable",
"KeySchema": [
{ "AttributeName": "Name", "KeyType": "HASH" },
{ "AttributeName": "Age", "KeyType": "RANGE" }
],
"AttributeDefinitions": [
{ "AttributeName": "Name", "AttributeType": "S" },
{ "AttributeName": "Age", "AttributeType": "S" }
],
"ProvisionedThroughput": {
"ReadCapacityUnits": 5,
"WriteCapacityUnits": 5
}
}$ aws dynamodb create-table --cli-input-json file://JSONSchemaDynamoDBTABLE.json
(3) 정책 생성 후 attach
다음으로는 아래의 정책을 갖는 IAM Policy를 생성합니다. 이 정책을 EKS Cluster의 역할 아래에 첨부해야합니다. 이 정책은 Kinesis로부터 DynamoDB로 쓸 때 필요한 action들을 포함하고 있습니다.
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "NifiaccesstoKinesis",
"Effect": "Allow",
"Action": [
"kinesis:DescribeStream",
"kinesis:GetRecords",
"kinesis:GetShardIterator",
"kinesis:ListStreams"
],
"Resource": "arn:aws:kinesis:us-west-2:{계정번호}:stream/kds-stream-nifi-on-EKS"
},
{
"Sid": "DynamoDBTableAccess",
"Effect": "Allow",
"Action": [
"dynamodb:BatchGetItem",
"dynamodb:BatchWriteItem",
"dynamodb:ConditionCheckItem",
"dynamodb:PutItem",
"dynamodb:DescribeTable",
"dynamodb:DeleteItem",
"dynamodb:GetItem",
"dynamodb:Scan",
"dynamodb:Query",
"dynamodb:UpdateItem"
],
"Resource": "arn:aws:dynamodb:us-west-2:{계정번호}:table/NifiStreamingTable"
}
]
}(4) 실습
본인의 도메인에 접속합니다. (https://{domain_name}/nifi)
이제부터 모든 flow 작성을 마우스로 해야합니다. 아래 그림과 같이 Process Group을 빈 배경으로 드래그합니다.
만든 Process Group을 더블클릭하면 다시 빈 화면이 나타납니다. 이제 이 배경에 예시 Flow를 직접 작성 (그리기?)합니다.
Processor를 빈 배경에 드래그 합니다. 그 후 아래와 같은 화면에서 kinesis를 검색하고 ConsumekinesisStream을 추가합니다.
만든 processor를 더블클릭 하고 아래와 같은 속성을 작성해줍니다.
a. Amazon Kinesis Stream Name
b. Application Name
c. Region
d. AWS Credentials Provider Service - Select AWSCredentialsProviderControllerService and create one.
사용할 수 있는 User의 Credential 정보도 입력해줍니다. (Access key, Secret Key)
위에서는 Kinesis Processor를 만들었는데요, 이번에는 destination인 DynamoDB Processor를 만들어봅니다. 똑같이 마우스로 Processor를 빈 배경에 드래그 하고 PutDynamoDBRecord로 셋팅하여 Amazon DynamoDB writer를 만들어줍니다. 아래와 같은 속성을 동일하게 셋팅합니다.
a. Record Reader - JSONTreeReader로 변경
b. AWS Credentials Provider Service - select the previously created configuration
c. Region
d. Table Name
e. Partition Key Field - select the partition field
이제는 Processor를 연결해줄 Connection을 생성해봅니다. Connection을 생성할 때는 Kinesis Processor로부터 DynamoDB Processor까지 이어줍니다.
이제 마무리 단계입니다. 아래 스크립트를 실행하여 Kinesis로 레코드들을 흘려줍니다.
import boto3
import json
import random
import time
kinesis_client = boto3.client('kinesis', region_name='us-west-2')
stream_name = 'kds-stream-nifi-on-EKS'
def create_sample_record():
names = ['Alice', 'Bob', 'Cindy', 'David', 'Ella']
locations = ['Seoul', 'Busan', 'Incheon', 'Daegu', 'Gwangju', 'Suwon', 'Dangjin', 'Guri']
record = {
'Name': random.choice(names),
'Age': str(random.randint(25, 40)),
'Location': random.choice(locations)
}
return record
def send_record_to_kinesis():
sample_record = create_sample_record()
partition_key = sample_record['Name']
response = kinesis_client.put_record(
StreamName=stream_name,
Data=json.dumps(sample_record),
PartitionKey=partition_key
)
print(f"Record sent to Kinesis: {sample_record}")
print(f"Response: {response}")
if __name__ == "__main__":
while True:
send_record_to_kinesis()
time.sleep(3)그럼에도 불구하고..
Apache NiFi를 이용하여 직접 파이프라인을 만들어보니 굉장히 감회가 새로웠습니다. 어린 시절에 그림판으로 그림 그렸던 것도 생각나기도 했고 생각보다 마우스를 활용해서 Flow를 셋팅하는게 쉽지는 않았던 것 같습니다. (심지어 컴포넌트들의 배치를 예쁘게 하고 싶은 욕심도 나는...) 굉장히 직관적이고 Streaming에 유용하다는 장점들이 있음에도 작업에 대한 트러블슈팅이 쉽지 않을 뿐더러 관리에도 어려움이 있고 오로지 UI를 통해서만 flow 작성 (그리기)을 할 수 있다는 것이 도입의 장벽이 될 것 같다고 느껴졌습니다.
저는 다음 시리즈로 다시 돌아오겠습니다! 감사합니다 (_ _)
