데이터 엔지니어링

aws summit 행사는, IT를 사용하는 산업의 기술 동향을 볼 수 있다. 왜냐면 여러 기업에서 aws 클라우드 환경으로 서비스를 운영 하기 때문이다.

최근에는 AI Agent 활용이 추세가 되면서, Google, MS Azure 를 멀티 클라우드 환경으로 혼용해서 쓰는 것 처럼보이지만, 서버, 스토리지 사용은 국내에서만큼은 여전히 aws 가 독보적인듯..

  

올해 행사는 2025.05.14~15 일 이틀간 이뤄졌고, 인상깊었던 세션 중 하나인 LG 전자의 발표를 소개하겠다.

  

주제는 "생성형 AI 기반 BI 플랫폼 인사이트 구현" 이다.

  

내용의 결론부터 말해보면 2가지 일 것 같다.

1) "Amazon Q in Quicksight 로 기존 대시보드의 한계를 극복했다".

2) "Multi AI Agent를 융합하여 구성한 Reporting 시스템 개발로 데이터를 보는 사용자의 관심을 끌었다"

  

그럼 배경과 함게 조금 더 자세한 결론을 얘기해보겠다.

  

일단 LG 전자에는 CDP(Customer Data Platform) 라는 데이터 플랫폼을 운영하고 있고, 셀프 BI 대시보드를 지향하는 환경을 사용자에게 제공하고 있었다.

LG 전자도 CDP 에 모인 데이터 기반으로 분석하는 과정은 어느 회사와 다르지 않다.

  

(1)문제 정의 -> (2)데이터 수집/추출 -> (3)데이터 정제/가공 -> (4)집계 /탐색 -> (5)데이터 해석 / 인사이트 도출 이 5가지 과정으로 분류 할 수 있다.

  

하지만 이 프로세스는 모두가 공감하는 문제를 갖고 있다. 비지니스 환경에서는 저 문제가 반복된다. 2번이 잘못되면 2~5번이 다시 수행되야 되는거지

그리고 현업이 요청 하면 대시보드를 제공하기 까지 리드 타임이 길다.

현업이 요청하면 한 번만 보고, 더 이상 대시보드를 보지 않는 경우도 있다. 만든 이후에는 관심도가 떨어지기 때문이다.

그리고 추가 분석항목들이 발생하면 제작자와 일정 줄다리기가 발생한다.

  

기능적으로도 문제가 있다. 고정된 차트 형태의 시각화를 제공해야 하는 제약이 있고 솔루션 자체의 기능 요청을 하면 답답함만 늘어난다. 글로벌 우선순위에 따라 진행되기 때문이지.

(실제로 LG 전자는 CDP 대시보드를 자체 구축했지만, 유지보수에 한계가 존재했다고 한다.)

  

그래서 Q in QuickSight 를 도입했고, BI 를 자연어 질의 기반으로 처리할 수 있는 환경을 갖추게 되었고, 현재는 전체 조직원이 활용하고 있다고 한다.

데이터 기반의 의사결정을 위한 생각의 시간을 줄여줬다 라는 표현을 했다.

  

플랫폼 아키텍쳐 측면에서 보면, ETL 한 DataMart 를 Amazon Redshift 로 이관하고, Quicksight 가 Datasource 로 연동되면서 SPICE 를 활용하는 구조다.

  

Quicksight 에 Amazon Q를 붙이기 전에, Amazon Bedrock으로 만든 자체 LLM 모델이 있었는데, 리소스가 많이 들고 복잡한 과정과 시간이 들어가서 Q를 사용했다고 했다.

생성형 AI 도 직접 만드는것보다 managed 서비스로 활용해보니 성능이 크게 다르지 않았다고 한다.

  

Q in Quicksight 와 태블로 대시보드의 성능 비교를 짧은 영상으로 보여줬는데 인상 적이었다.

  

그 다음에는 Multi Agent Report 시스템을 소개했다.

LG 전자가 생각하는 "대시보드 기반의 분석의 한계" 는 다음과 같다

(1) 분석에 시간이 많이 든다.

(2) 너무 많은 차트는 복잡도를 높인다. 그렇다고 차트를 줄이기도 그렇고

(3) 단순한 질문 & 답변으로는 종합적인 인사이트 도출이 어렵다.

(4) 현업들은 정형 데이터에서 정확성을 얻기 쉽지 않다.

  

다각도의 데이터 자동 분석이 필요했고, 그래서 만든게 Multi Agent Report 시스템이라고 한다.

제공된 서비스를 보면 Prompt 로 데이터셋과, 스키마 정보와 함께, 비지니스 요구사항을 자세히 작성해서 주문하면, 컨텍스트를 분석하는 Agent, 보고서 목차를 정리하는 Agent, 분석 코드를 만들고 실행하는 Agent , 결과를 취합하고 시각화하는 Agent, 보고서를 발송하는 Agent 이렇게 여러 Agent 가 유기적으로 동작하는 모습을 보였다.

  

머니테터 같은 10장 내외의 레포트가 자동으로 만들어지고, 이메일로 발송된다. LG 전자는 이 서비스로 보고서 작성 시간을 단축하고 의사결정을 빨리 할 수 있기를 기대하고 있다고 했다.

여기까지가 LG 전자 서밋의 내용 요약인데, 발표 자료 안에 있던 모든 기획과 구성들을 어느 기간안에 진행했는지, 그리고 데이터 처리 규모와 배치/스트리밍 데이터의 흐름 등이 궁금했다.

LG 전자... 데이터도 잘 하네..

끝.

1.N8N 소개

아래 포스팅을 참고하자.

https://jssvs.tistory.com/110

주요 특징

• 노드 기반 아키텍처: 직관적인 시각적 인터페이스를 통해 노드를 연결하여 워크플로우 구성
• Fair-code 라이선스: 대부분의 사용 사례에서 무료로 사용 가능
• Self-hosted 가능: 클라우드 또는 온프레미스 환경에서 직접 호스팅 가능
• 700개 이상의 통합: Slack, Google Sheets, Airtable, Notion 등 다양한 서비스와 연동
• 커스텀 노드 개발: JavaScript/TypeScript로 사용자 정의 노드 개발 가능
• 워크플로우 자동화: 스케줄링, 웹훅, API 호출 등을 통한 자동화 트리거 지원

2.Helm Chart를 이용한 n8n 설치 과정

A) Helm 리포지토리 추가

$ helm repo add n8n https://n8n-community.github.io/n8n-helm
$ helm repo update

B) chart value 작성

service:
  type: NodePort
  port: 5678
  name: http
  annotations: {}


main:
  extraEnvVars:
    N8N_SECURE_COOKIE: false

C) N8N 배포 

#!/bin/bash
helm repo add community-charts https://community-charts.github.io/helm-charts
helm repo update


echo "install n8n by helm.."

VERSION="1.5.11"
helm upgrade --cleanup-on-fail \
  --install my-n8n community-charts/n8n \
  --namespace n8n \
  --create-namespace \
  --version $VERSION \
  --values values.yaml


sleep 1

echo "install done.."

D) 외부 접속을 위한 ingress 생성

apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: n8n-ingress
  namespace: n8n
  annotations:
    alb.ingress.kubernetes.io/scheme: internet-facing #public 오픈 시
    alb.ingress.kubernetes.io/target-type: instance  
    alb.ingress.kubernetes.io/subnets: [서브넷 리소스]
    alb.ingress.kubernetes.io/security-groups: [보안그룹 리소스]
    alb.ingress.kubernetes.io/load-balancer-name: alb-dev-n8n

spec:
  ingressClassName: alb
  rules:
    - http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: my-n8n
                port:
                  number: 5678

1.N8N 소개와 개요

N8N은 "Node-RED"나 "Zapier"처럼 워크플로우 자동화를 위한 툴이다.

가장 큰 특징은 오픈소스라는 점이다. MIT 라이선스를 기반으로 누구나 자유롭게 사용할 수 있고, 클라우드뿐만 아니라 자체 서버에 설치해서 운영할 수도 있다. 특히 IT 개발자나 엔지니어가 업무 자동화, 데이터 처리, 시스템 연동 등을 구현할 때 유용하다.

N8N은 다양한 API와 연동이 가능하고, 직관적인 UI 덕분에 비전문가도 기본적인 워크플로우를 쉽게 구성할 수 있다. 예를 들어, 블로그 운영자는 새로운 글이 발행되면 자동으로 트위터에 공유하거나, RSS 피드에서 콘텐츠를 가져와 블로그에 자동 업로드하는 등 블로그 자동화에 활용할 수 있다.

요약하면, N8N은 누구나 자유롭게 사용할 수 있는 오픈소스 워크플로우 자동화 툴이며, 기술 스택과 업무 방식에 맞춰 유연하게 확장 가능한 것이 강점이다.

2.N8N 설치 방법 소개

N8N은 로컬 환경, Docker, 클라우드 서비스 등 다양한 방식으로 설치할 수 있다. 여기서는 가장 많이 쓰이는 Docker 기반 설치 방법을 중심으로 설명한다.

A. 사전 준비 사항

• Docker 설치 (https://docs.docker.com/get-docker/)
• Docker Compose 설치
• 최소 1GB 이상의 RAM 권장

B. Docker Compose 파일 작성

아래는 기본적인 docker-compose.yml 예시이다:

version: '3'
services:
  n8n:
    image: n8nio/n8n
    restart: always
    ports:
      - "5678:5678"
    volumes:
      - ~/.n8n:/home/node/.n8n
    environment:
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=admin
      - N8N_BASIC_AUTH_PASSWORD=adminpassword

C. 실행 명령어

docker-compose up -d

이제 브라우저에서 http://localhost:5678로 접속하면 로그인 후 N8N 대시보드에 들어갈 수 있다.

Docker 외에도 다음과 같은 방법으로 설치할 수 있다:

• npm을 이용한 전통적 설치
• N8N 클라우드(공식 유료 서비스)
• Heroku, Render 등의 PaaS 이용

오픈소스의 장점을 살려서 원하는 방식으로 자유롭게 설치 환경을 구성할 수 있는 것이 N8N의 매력이다.

docker $ volume create n8n_data
docker $ run -it --rm --name n8n -p 5678:5678 -v n8n_data:/home/node/.n8n docker.n8n.io/n8nio/n8n

3.결론

N8N은 강력한 오픈소스 워크플로우 자동화 툴이다. 개발자나 엔지니어가 시스템 통합과 업무 자동화를 손쉽게 구현할 수 있도록 도와준다. 특히 블로그 자동화에 활용하면 반복적인 작업을 줄이고 효율적인 운영이 가능하다.

설치와 사용이 비교적 간단하고, 다양한 API와 연동할 수 있기 때문에 어떤 기술 환경에서도 잘 맞는다. N8N을 활용해서 자신만의 스마트한 작업 흐름을 구성해보자.

apache spark 은 데이터를 다루는 사람들에게는 필수적으로 알아야 하는 기술이다.

내 방식대로 정의하자면 spark 은 framework 이면서, 빅 데이터 처리 엔진이라고 할 수 있을 것 같다.

요즘엔 Snowflake 같은 SaaS 로 SQL Like 기반의 데이터 처리도 많이 하지만, 오픈소스 진영에서는 무조건 spark 이다.

아래 예전에 정리했던 기술을 다시 보면서, 블로그 포스팅을 해본다

1.스파크에 대해

• 하나의 서버에서 처리할 수 있는 수준 이상의 대용량 데이터 처리를 가능하게 해준다
• 고수준 API 를 제공한다
• 동종 시스템 중 가장 빠른 축에 속한다.

스파크 버전 규칙

• [메이저].[마이너].[유지보수]

왜 스칼라인가

• 스파크는 스칼라로 쓰여져 있다.
• 스파크는 함수형 프레임워크이며, 불변성이나 람다 정의 같은 개념에 크게 의존하고 있기 때문에 함수형 프로그래밍 언어인 스칼라가 좋다
• 자바 API 보다 훨씬 사용하기 쉽다
• JVM 과의 통신 비용이 랭기지중 가장 좋다. (객체 변환에 드는 비용정도)

2.스파크 동작 개념

스파크의 독보적 장점

• 메모리 기반 처리
• 지연 평가 ( lazy operation )

스파크 위치

• JVM 위에서 연산을 수행하는 것일 뿐, 데이터 저장 솔루션은 아니다
• 클러스터 매니저 종류 : 단독 클러스터매니저(StandAlone Cluster Manager), 아파치 메소스, 하둡 얀

스파크 컴포넌트

• 스파크 코어 - Java, Scala, Python, R 로 API 제, RDD 와 RDD API, JVM 사이에 데이터를 읽고 쓰는 I/O 제공
• 스파크 SQL - DataFrame ( 반구조화의 데이터 타입을 위한 인터페이스) 제공
• 스파크 스트리밍 등

병렬 연산 모델 : RDD

• 스파크의 드라이버 혹은 마스터 노드를 위한 프로그램을 사용자가 만들어야 한다
• RDD는 익스큐터 혹은 슬레이브 노드에 저장된다
• RDD를 구성하는 객체는 파티션이라 하며, 경우에 따라 다른 노드에서 계산될 수 있다.
• 클러스터 매니저는 애플리케이션에서 설정한 파라미터에 따라 익스큐터를 실행하고 분산해 주는 역할을 한다
• 드라이버가 RDD 데이터의 처리 계획을 결정하면 실행을 지연하고, 최종 RDD를 계산해야 하는 시점에 실행된다.( 보통 저장 장치에 써야 할 시점이나 결과 데이터를 드라이버에 보내야 할 때)

지연평가

• RDD의 연산 종류는 transformation 과 action 이 있다.
• 액션은 데이터를 드라이버로 되돌려주든지(count, collect 등) 데이터를 외부 저장 시스템에 쓰는것(CopyToHadoop) 등의 일이다.
• 액션은 스케쥴러를 시작하게 하며, RDD 트랜스포메이션 간의 종속성에 바탕을 둔 DAG 를 생성한다.
• 트랜스포메이션의 로직 에러가 발생할 때, 지연 평가 때문에, 액션이 수행된 지점에서 실패한 것으로 나타나는 경우에 유의하자
  • word count 프로그램에서 null pointer exception 이 발생한다고 가정할때, 코드가 contains를 체크하는 시점이 아니라 collect 단계에서 예외가 발생한다.

메모리 영속화 & 메모리 관리

• 맵리듀스와 비교해 스파크 성능상 이점은 반복 연산이 들어있는 사례이다.
• 스파크가 처리하는 데이터를 디스크에 기록하지 않고 익스큐터 메모리에 데이터를 로드해 놓는 것이다.
• 메모리 관리의 3가지 옵션
  1. 메모리에 직렬화되지 않은 자바 객체 - RDD 객체를 직렬화 하지 않고 그대로 저장한다. ( 직렬화 비용이 안드는 대신 메모리 공간 사용이 비 효율이다. )
  2. 메모리에 직렬화된 데이터 - RDD 객체를 네트워크 전송이 가능한 바이트 스트림으로 변환한다. ( 데이터를 읽는데 CPU 가 더 많이 사용되므로 접근방식은 더 느리지만 메모리 공간 사용 측면에서 효율적이다. 크리오(Kryo) 직렬화를 쓰면 공간 측면에서도 효과적이다.)
  3. 디스크 - 익스큐터 램에 담기에 파티션이 큰 RDD 라면 디스크에 데이터를 쓸 수 있다. ( 반복 연산시 속도 면에서 불리하지만 장애에 안전하다.)
• persist() 의 기본 옵션은 RDD를 메모리에 직렬화되지 않은 상태로 저장한다.

불변성과 RDD 인터페이스

• RDD는 정적인 타입인 데다 불변한 성격을 가지고 있어, Transformation 을 호출하는 것이 새롭게 정의한 속성들을 가진 새로운 RDD를 리턴하는 행위다.
• RDD 생성 방식
  • 기존 RDD에 Transformation 호출
  • SparkContext 객체로부터 생성
  • DataSet이나 DataFrame을 변환 ( 이것들은 SparkSession 으로 부터 만들어짐)
• SparkContext는 스파크 클러스터와 실행중인 스파크 애플리케이션 하나와의 연결을 나타낸다.
• RDD 속성을 알 수 있는 함수
  • partitions() - 분산 데이터 셋의 부분들을 구성하는 파티션 객체들의 배열을 리턴한다. getPartition()의 결괏값과 같다
  • iterator(p,parentIters) - 각각의 부모 파티션을 순회하는 반복자가 주어지면 파티션 p의 구성요소들을 계산해낸다.
  • dependencies() - 종속성 객체의 목록을 리턴한다. 스케쥴러가 현재의 RDD가 어떤식으로 다른 RDD에 종속될지 알려준다.
  • partitioner() - element 와 partition 사이에 연관되는 함수를 갖고 있는 RDD라면 스칼라의 option 타입으로 partitioner 객체를 리턴한다.
  • perferredLocations(p) - 파티션 p의 데이터 지역성에 대한 정보를 리턴한다. p가 저장된 각 노드의 정보를 문자열로 표현한 시퀀스를 리턴한다

RDD의 종류

• RDD는 정적인 타입인 데다 불변한 성격을 가지고 있어, Transformation 을 호출하는 것이 새롭게 정의한 속성들을 가진 새로운 RDD를 리턴하는 행위다.

넓은 종속성 vs 좁은 종속성

• 종속성이 넓으냐 좁으냐는 트랜스포메이션 평가에 중요한 영향을 끼치며 성능에도 크게 작용한다.
• 좁은 종속성
  • 자식 RDD 의 각 파티션이 부모 RDD의 파티션들에 대해 단순하고 한정적인 종속성을 갖는다
  • 부모가 최대 하나의 자식파티션을 갖는 경우
  • map, filter, mapPartitions 등의 연산이 이 패턴을 따른다.
• 넓은 종속성
  • 자식 RDD가 다수의 부모 RDD의 파티션들과 관계를 맺고 있는 경우
  • groupbykey, sort, reducebykey 등과 같이 Shuffle 을 일으키는 함수가 이 패턴을 따른다.
  • 셔플 비용이 가장 크다.
• map 은 파티션 간 이동이 없는 연산, coalesce 는 파티션을 합치는 연산으로 파티션 개수를 줄이는 목적의 함수이다.
• join 함수는 두개의 부모 RDD 가 어떻게 파티션되었는지에 따라 좁거나 넓은 종속성을 가질 수 있다.

스파크 잡 스케쥴링

• 잡 실행 과정
  • 스파크 프로그램 자체는 드라이버 노드에서 실행되며 일련의 명령들을 익스큐터에게 보낸다.
  • 애플리케이션들은 클러스터 매니저에 의해 스케쥴링되고, 각각 하나의 SparkContext를 가진다.
  • 스파크 애플리케이션들은 공존하는 여러 개의 잡을 차례로 실행할 수 있다.
  • Job들은 애플리케이션의 한 RDD가 호출하는 각 액션에 대응한다.
• 자원할당
  • 정적할당과 동적 할당이 가능
• 스파크 애플리케이션
  • 잡들은 드라이버 프로그램의 SparkContext에 정의되어 있다.
  • SparkContext가 생기면 스파크 애플리케이션이 구동한다.
  • SparkContext를 실행하면 드라이버와 익스큐터들이 각 작업 노드에서 구동된다.
  • 각 익스큐터는 JVM을 가지며 한 노드에 여러 개의 익스큐터가 존재할 수 있다.
  • 스파크 잡이 실행될 때 익스큐터는 RDD를 계산할 태스크 실행을 위한 슬롯을 가진다.
• 스파크 잡 해부
  • 각 액션마다 스파크 스케쥴러는 실행 그래프를 만들고 스파크 잡을 실행한다.
  • 각 잡은 최종 RDD를 만들어 내는데 필요한 데이터 변환의 각 단계를 의미하는 스테이지(Stage)들로 구성된다.
  • 각 스테이지는 각 병렬 연산의 한 단위를 의미하며 익스큐터들 위에서 실행되는 다수의 태스크(Task)들로 구성된다.
  • Spark Aplication -> 잡 -> 스테이지1 & 스테이지2 ... -> 태스크 1& 태스크 2...
  • SparkContext / SparkSession -> 액션 -> 넓은 포메이션 -> 하나의 파티션 평가를 위한 연산
• 잡
  • 스파크 실행 구성에서 가장 높은 단계
  • 하나의 잡 = 하나의 액션에 대응하며, 액션은 스파크 애플리케이션의 드라이버의 프로그램에서 호출한다.
• 스테이지
  • 잡의 부분들을 스테이지로 정의한다.
  • 넓은 트랜스포메이션에 의해 생성되는 셔플 의존성에 대응한다.
  • 하나의 스테이지는 다른 익스큐터나 드라이버와의 통신 없이 계산 가능한 태스크들의 집합으로 생각할 수 있다.
• 태스크
  • 하나의 스테이지는 여러 개의 태스크로 이루어진다.
  • 실행 계층에서 가장 작은 단위이며, 익스큐터가 태스크 실행을 위해 동적으로 할당된 여러개의 슬롯을 가진다.
  • 스테이지당 태스크의 개수는 해당 스테이지의 결과 RDD의 파티션 개수에 대응된다.
  • 익스큐터 코어 개수의 총합 = 익스큐터당 코어의 개수 X 익스큐터 개수를 쓰면 sparkConf로 부터 동시 실행의 태스크 개수를 구할 수 있다.
  • 태스크 분산 과정은 TaskScheduler 가 담당하는데, 페어 스케쥴러인지 FIFO 스케쥴러인지 등에 따라 다르다.

쿠버네티스 복습을 하면서 스토리지 관련해 요약했던 내용을 포스팅 해보겠다.

1.EBS란 ?

• Elastic Block Store (EBS)
• EC2 에서 쉽게 사용할 수 있는 영구적인 Volume (linux 가 볼륨을 디바이스로 인식하는)
• mkfs 명령어를 사용하여 mount 가능
• 다양한 유형
  • ssd 기반 - gp2
  • iops 속성이 존재. io1, io2., st1, sc1 등.
  • iops 조절이 가능한 gp3

2.PV란 ?

• Persistent Volume
• 물리적인 볼륨을 나타냄
• Cluster 단위의 resource (node 처럼)
• 관리자가 생성하거나 pvc 로 부터 동적으로 생성됨
• 예를 들어 pv 는 요리사가 만든 피자 한 판을 굽는 것이다.

3.PVC란?

• namespace 단위의 resource (pod 처럼)
• 사용할 용량, access mode, 등의 속성을 갖고 있음
• PV의 요청을 의미하는 추상 오브젝트
• 예를 들어 pvc 는 피자 한 판에서 피자 한 조각을 손님이 접시에 담아 가져오는 것이다.

4.provisioner 란?

• storageclass 의 설정과 pvc 의 설정을 읽어 aws ebs 등 물리적인 volume 과 pv obeject 생성을 담당.

5.csi Driver 란?

• Container Storage Interface Driver
• EKS 에서는 쿠버네티스에 내장된 provisioner 를 모두 삭제하고 별도의 컨트롤러 파드를 통해 동적으로 프로비저닝을 사용할 수 있도록 만들었음. 그것이 바로 csi driver

아래 실습을 간단히 해본다

먼저 스토리지 클래스를 만들고

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: my-gp2
provisioner: kubernetes.io/aws-ebs
parameters:
  type: gp2
  fstype: ext4

PV를 만든다.

apiVersion: v1
kind: PersistentVolume
metadata:
  name: my-pv
spec:
  capacity:
    storage: 10Gi
  accessModes:
    - RewadWriteMany
  persistentVolumeReclaimPolicy: Retain
  storageClassName: my-gp2
hostPath:
    path: /mnt/data


---

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: my-pvc-ebs
  namespace: nextcloud

spec:
  storageClassName: my-gp2
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 10Gi
  persistentVolumeReclaimPolicy: Retain

간단하게 Pod 를 만들어 마운트해보자

apiVersion: v1
kind: Pod
metadata:
  name: task-pv-pod
spec:
  nodeSelector:
    kubernetes.io/hostname: kube-01
  volumes:
    - name: task-pv-storage
      persistentVolumeClaim:
        claimName: task-pv-claim
  containers:
    - name: task-pv-container
      image: nginx
      ports:
        - containerPort: 80
          name: "http-server"
      volumeMounts:
        - mountPath: "/usr/share/nginx/html"
          name: task-pv-storage

카프카 구축 전에 예전에 정리했던 내용 기반으로 이론을 다시 정리해봤다. 

요즘에는 운영을 쉽게 하려고 MSK, 컨플루언트 카프카를 많이 사용 하는 것 같다.

아무튼 본론으로..

1.카프카란?

• 분산 이벤트 큐.
• 분산 이벤트 스트리밍 플랫폼
• 카프카 컨슈머, 프로듀서, 스트림즈, 커넥트 등 연동 API 제공
• 초당 수백만개 데이터를 처리할수 있으므로 빅데이터에 적합
• 분산 데이터를 통해 24시간 365일 안전하게 데이터를 처리할 수 있는 고가용성 기능 제공

2.왜 카프카?

• 고가용성
  • 서비스에 지장없는 운영을 보장.
• 낮은 지연
• 확장성
• 높은 처리량
  • 높은 처리량을 감당하지 못한다면, 서비스를 유지하기 힘듦

RabbitMQ , Redis 와의 차이점

• 이벤트 브로커
  • 서비스에서 발생한 이벤트를 브로커의 큐에 저장함
  • 딱 한번 일어난 이벤트 데이터를 브로커에 저장함으로써 단일 진실 공급원으로 사용 및 재처리가 가능
  • 마이크로 서비스 아키텍쳐에서 중요한 역할을 함
  • kafka , kinesis
• 메세지 브로커
  • 대규모 메세지 기반 미들웨어 아키텍쳐에서 사용
  • RabbitMQ, Redis

카프카 구조

• 기존 1:1 매칭으로 개발하고 운영하던 데이터 파이프라인은 커플링으로 인해 한쪽 이슈가 생기면 다른 한쪽에도 영향이 간다. → 카프카는 이러한 의존도를 타파하였다. (디커플링)
• 큐에 데이터를 보내는 것이 프로듀서이고 큐에서 데이터를 가져가는 것이 컨슈머다

카프카 특징

• 높은 처리량
  • 높은 처리량을 감당하지 못한다면, 서비스를 유지하기 힘듦
  • 우리 비지니스의 성공여부는 어떤 Threash hold 에 걸쳐지면 안된다.
  • 파티션 단위를 통해 동일 목적의 데이터를 여러 파티션에 분배하고, 이런 파티션을 컨슈머로 병렬처리할수 있는것이 큰 특징
  • 파티션 개슈만큼 컨슈머 개슈를 늘릴수 있다
• 확장성
• 영속성
  • 파일 io 성능 향상을 위해 os 에서 담당하는 페이지 캐시를 이용한다. 그래서 파일을 쓰고 읽는데도 빠를수 있다.
• 고가용성

3.카프카 구성 요소

토픽

• 구체화된 이벤트 스트림 = 쉽게 큐로 이해하면 됨
• 하나의 토픽에 여러 Producer / Consumer 가 존재할 수 있다.
• 토픽은 담는 데이터에 따라 이름을 줄 수 있다.

컨슈머

• 기본적으로 가장 오래된 순서대로 가져감 - 0번 오프셋부터
• 새로운 컨슈머가 구독을 하게 되도 가장 오래된 순서대로 가져감
  • auto.offset.reset = earliest 인경우

파티션

• 카프카의 토픽들은 여러 파티션으로 나눠짐.
• 파티션의 끝에서 0번 부터 차곡차곡 쌓이게 됨
• 토픽 = 논리적인 개념이라면, 파티션은 물리적인 저장소에 저장하는 단위
• 각 파티션은 Append-only 방식으로 기록됨
• 특정 파티션으로 데이터를 쓸수 있고, 명시되있지 않으면 RoundRobin 방식으로 파티션을 배정한다
• 파티션을 늘린다면?
  • 파티션을 다시 줄일수는 없다.
  • 컨슈머 개수가 늘어날때 분산 처리할 수 있다.
  • 신규 데이터는 2개의 파티션 중어디로 들어갈까?
  • 보통은 라운드로빈으로 파티션을 할당함
  • 키의 해시값으로
• 파티션 삭제 주기는?
  • log.retention.ms : 최대 record 보존 시간
  • log.retension.byte : 최대 record 보존 크기

오프셋

• 각각 파티션의 레코드는 Offset 식별자 정보를 가짐, 데이터 번호
• 카프카는 메세지 순서를 보장 하지 않음. 하지만 파티션이 1개라면 보장할지도?

4.카프카 클러스터

카프카 클러스터

• 카프카 브로커
  • 설치된 카프카의 서버 단위
  • 보통은 3대로 구성
• replication
  • replication 이 1인 상태라면 파티션이 브로커 서버에 1개만 저장된다.
  • replicaion 이 2라면 원본 하나, 복제본 1개의 파티션이 각각의 브로커 서버에 저장된다.
  • 따라서 replication 개수 ≤ 브로커 서버 개수
  • 원본 파티션 = Leader 파티션, 복제본 파티션 = follow 파티션
  • replication 의 설정된 값에 따라 서로 다른 브로커 서버에 파티션의 복제본이 생긴다.
• ISR(In-SyncReplica)
  • 리터 파티션의 레코드 개수 만큼 팔로워 파티션의 개수가 동일하게 복제가 된 안정된 상태
• ACK
  • 카프카 프로듀서는 ack 를 이용해 고 가용성 보장
    • ack = 0 , response 무시. 속도는 빠르지만 유실이 있음.
    • ack = 1, reponse 를 받음, 파티션 복제는 보장 못함. 유실 가능성 있음
    • ack = all , response 받고, follwer partition 저장확인 절차를 거침. 유실 가능성 없음
  • replication 개수가 늘어난다면 성능 저하.
• 파티셔너
  • 데이터를 토픽에 어떤 파티션에 넣는지 결정하는 역할을 함
  • 메세지 키 또는 메세지 값에 따라 파티션이 결정됨
  • hash(키) = 파티션 넘버
• 카프카 lag
  • 운영시에는 consumer lag이 발생
  • lag 이란 = 컨슈머가 마지막으로 읽은 offset - 프로듀서가 마지막으로 넣은 offset
  • 한개의 토픽과 컨슈머 그룹에 대한 lag 이 여러개 존재 하게 된다.
  • max lag 에 대한 모니터링이 운영시에는 필요하다
• lag burrow
  • golang 으로 개발된 오픈 소스
  • 컨슈머 lag 모니터링을 도와주는 독립적인 애플리케이션
  • 멀티 카프카 클러스터 지원
    • 2개이상의 카프카 클러스터를 운영할때, 하나의 burrow 로 운영 가능
• 주키퍼
  • 코디네이션 애플리케이션
  • 브로커 서버와 통신하며 상태관리, 컨슈머와의 통신, 카프카 메타데이터 정보를 저장함.

빅데이터 처리 플랫폼을 EMR on EKS 로 도입하면서, 쿠버네티스 운영도 함께 하고 있다.

요즘엔 Spark 개발보다는 쿠버네티스 엔지니어링을 더 많이 하는 것 같아서, 정체성의 혼란이 올 때도 있지만..

나의 본질은 데이터 엔지니어링에 있다고 믿는다.

지난번에 한 번 정리한 글이 있는데, 새로운 명령어와 함께 다시 올려본다.

argoCD 사용자는 kubectl 을 쓸 일이 거의 없겠지만, 난 두개를 다 쓰고 있는 입장에서 kubectl 로 리소스를 직접 다루는게 더 재미있는

편이다.  아무튼.. 누군가에게 도움이 되길 바라며

1.  리소스 상태 및 로깅 관련

• 쿠버네티스 클러스터에서 발생하는 이벤트 로깅, pod , Node 의 리소스 상태 변화 순서로 확인하는게 좋다.

기본 이벤트 확인
$ kubectl get events

특정 네임스페이스 이벤트 확인
$ kubectl get events -n <namespace>

실시간 이벤트 감시
$ kubectl get events --watch

출력 형식 지정
$ kubectl get events -o wide


특정 Pod 상세 정보 확인
$ kubectl describe pod <pod-name>

네임스페이스 지정
$ kubectl describe pod <pod-name> -n <namespace>

모든 Pod 정보 확인
$ kubectl describe pod



Pod Logs (Pod 로그 확인)
$ kubectl logs <pod-name>

특정 컨테이너 로그 확인
$ kubectl logs <pod-name> -c <container-name>

실시간 로그 스트리밍
$ kubectl logs <pod-name> -f


시간별 로그 필터링
$ kubectl logs <pod-name> --since=1h



기본 노드 목록
$ kubectl get nodes

자세한 정보 포함
$ kubectl get nodes -o wide

JSON/YAML 형식으로 출력
$ kubectl get nodes -o json

노드 내 모든 리소스 정보 확인
$ kubectl describe node

2. 헬름 차트 관련 (helm chart) 

• 헬름을 이용한 서비스 설치도 많이 하게 되는데, 기본 개념을 정리하면 다음과 같다
  • 헬름 (helm) : 쿠버네티스 패키지 매니저
  • 차트 (chart) : 쿠버네티스 리소스를 하나로 묶은 템플릿 (yaml 의 묶음)
• 헬름을 이용해 차트 기반으로 서비스를 배포할때 다음 용어도 알아두면 좋다
  • 릴리즈 : 특정 차트를 클러스터에 설치한 인스턴스
  • 레포지토리: 차트를 저장하고 배포하는 저장소
• 순서는 사용자가 정의해둔 차트에 value 값을 custom 하게 설정하고, 레포지토리를 추가한 다음. 릴리즈 버전과 함께 설치하면 된다.

레포지토리 추가
$ helm repo add <repo-name> <repo-url>

레포지토리 목록 확인
$ helm repo list

레포지토리 업데이트
$ helm repo update

차트 검색
$ helm search repo <keyword>

기본 차트 설치
$ helm install <release-name> <chart-name>

특정 네임스페이스에 설치
$ helm install <release-name> <chart-name> -n <namespace>

value 파일로 커스터마이징
$ helm install <release-name> <chart-name> -f values.yaml

직접 값 지정으로 설치
$ helm install <release-name> <chart-name> --set key1=value1,key2=value2


설치된 릴리즈 확인
$ helm list


차트 업그레이드
$ helm upgrade <release-name> <chart-name>

릴리즈 히스토리 확인
$ helm history <release-name>

롤백 
$ helm rollback <release-name> <revision>

릴리즈 삭제
$ helm uninstall <release-name>

helm chart 조회는 아래에서 하면 된다.

https://artifacthub.io/

빅데이터 환경에서 다양한 애플리케이션, 플랫폼 도구들을 실험할 때 데이터는 항상 필요하다.

실제 운영 환경의 데이터를 활용하면 좋겠지만, 보통은 개발/스테이징/운영의 환경을 나눠놓기 때문에, 데이터도 개발용 데이터가 필요할 때가 있다.

아래 간단하게 yaml 포맷으로 데이터 스키마를 정의하면, 명령어 실행으로 데이터를 출력해주는 간단한 툴을 만들었다.

누구나 가져다 쓸 수 있고, 수정해서 쓸 수 있다.

사실 이미 공개된 오픈 소스 제네레이터는 많지만, 파티셔닝된 데이터출력이라던지 데이터 포맷을 스스로 정의해줘야 하는 특수한 경우를 고려해서 그냥 만들어봤다.

만들어볼 줄 도 알아야 하니까.

1. BigdataSimpleGenerator 소스코드 

https://github.com/jaysooo/bigdata-simple-generator

2. 빠른 시작

A.데이터 스키마 정의하기

• 기본적으로 int, string, float( 자리수 포함), 범위, 옵션 리스트 를 정의할 수 있다.
• S3 , 로컬을 지원한다.
• 레코드 개수를 정의할 수 있다.

data_spec:
  table_name: stg-event-dummy
  records: 1000
  file_format: csv # csv, parquet, json
  file_prefix: data
  destination: /YourPath/sample_data
  # destination: s3://yourbucket/event-data/
  partition_by:
    name: partition_key
    range:
      min: "2025-01-01"
      max: "2025-01-10"
  table_schema:
    - name: id
      type: int
      prefix: id_
    - name: device_name
      type: string
      prefix: device_
    - name: device_type
      type: string
      range:
        items:
          - "iphone"
          - "galaxy"
          - "xiaomi"
          - "huawei"
    - name: event_time
      type: timestamp
    - name: event_type
      type: string
    - name: event_rate
      type: float
      range:
        min: 0 
        max: 1
      decimal_point: 2

B.실행하기

$ python data-generator.py --config config.yaml --producer pyarrow

3. 출력 데이터

4. 참고

• pyarrow dataframe 으로 데이터를 생산하기 때문에, 데이터량에 따라 머신의 메모리 가용 공간을 확인한다.
• 인터페이스만 구현한다면 데이터 출력 모듈을 직접 구현해서 출력할 수 있다. (pyspark, pandas 등 )

오랜만에 포스팅을 하는 것 같다.

오늘은 GPT 를 이용해 간단한 동영상 쇼츠 제작 방법을 소개해보려고 한다.

여러 방법들이 있을 거고 정답은 없는 것 같다. 나는 X에서 최근 공개한 Grok 과 openAI 사의 ChatGPT 를 함께 이용해서 만들었다.

절차부터 얘기하면 다음과 같다.

1) grok 을 이용해 대본만들기

2) chatGPT 의 영상 제작 모델에 대본을 프롬프트로 입력해 영상 제작을 의뢰한다.

Grok (https://x.ai/) 으로 브라우저를 옮겼다. 그리고 나는 대본을 만들어달라는 프롬프트를 아래 처럼 주문했다.

답변을 복사하고 chatgpt (https://chatgpt.com/) 로 브라우저를 넘겨보자.

GPT 탐색 을 눌러보면 video maker 와 관련된 여러 모델들을 검색할 수 있는데, 나는 1위에 랭크되있는 모델을 택했다.

나는 데이터엔지니어기 때문에, apache airflow 를 30초 로 소개하는 영상을 주문해보았다.

대본을 프롬프트에 입력하고 영상 제작을 주문하면 연계된 서비스 페이지로 링크를 하나 보내줄 것이다.

링크를 타고 들어가면, 영상을 제작하는 과정을 조금 기다려야 한다.

물론 !!! 자동으로 생성된 영상은 한 20% 정도 부족해보인다.

그래서 동영상을 편집하는 모드에서 자막과 영상을 업로드할 수도 있다. (여기서 노가다가 시작되지 않을까.. ) 

퀄리티가 아주 만족할 만한 수준은 아니지만, 그래도 짧은 시간에 제작한 것 치곤 괜찮다.

AI 가 제작해준 영상을 아래 업로드하면서 마무리한다.

(아.. 용량 제한으로 영상 업로드가 안되네.. ) 

참고) 프롬프트

**** GROK Prompt ****

이제부터 너 유튜브 대본을 작성하는 작가이자 데이터 엔지니어인 제인이야 

난 한국에서 유튜브 채널을 만들어 영상을 올릴 계획이야. 

이번에는 apache airflow 를 소개하는 영상을 만들거야

다음 요구사항을 만족하도록 대본과 정보를 만들어줘

- 영상 시간은 30초야
- airflow 의 소개와, 기능 그리고 장점으로 구성해줘
- airflow 를 사용자에게 홍보하는 느낌으로 만들어줘



**** ChatGPT Prompt ****

이제부터 너 유튜브 대본을 작성하는 작가이자 데이터 엔지니어인 제인이야

난 한국에서 유튜브 채널을 만들어 영상을 올릴 계획이야.

이번에는 apache airflow 를 소개하는 영상을 만들거야

다음 대본을 줄테니 영상을 제작해줘

대본 (30초)

"안녕하세요! 오늘은 데이터 엔지니어링의 필수 도구, Apache Airflow를 소개해드릴게요. Airflow는 복잡한 데이터 파이프라인을 쉽게 관리하고 스케줄링할 수 있는 오픈소스 플랫폼이에요. DAG를 통해 작업의 의존성과 순서를 명확히 정의하고, AWS, Google Cloud 같은 서비스와도 간편하게 연동할 수 있죠. 확장성이 뛰어나고, 활발한 커뮤니티 지원으로 계속 발전 중이에요. 게다가 직관적인 UI로 워크플로우를 한눈에 모니터링할 수 있어요. 데이터 작업을 효율적으로 관리하고 싶다면, Airflow를 꼭 사용해보세요!"

정보 요약

소개: Apache Airflow는 데이터 파이프라인을 관리하고 스케줄링하는 오픈소스 플랫폼입니다.

기능: DAG(Directed Acyclic Graph)로 작업 흐름을 정의하고, 다양한 서비스(AWS, Google Cloud 등)와 유연하게 통합됩니다.

장점: 확장성이 뛰어나고, 커뮤니티 지원이 활발하며, UI로 워크플로우를 시각적으로 관리할 수 있어 사용자 친화적입니다.

1. Yunikorn 이란 ?

• 쿠버네티스에서 동작하는 Batch , Data & ML 서비스를 위한 리소스의 파워를 unleash(해방,촉발시키는) 하는 스케쥴러다.(1)
• k8s 기본 스케쥴러의 대안으로 활용될 수 있는 스케쥴러로, 특히 다양한 리소스의 요구사항을 만족시킬 수 있는 기능들을 제공한다.

2.왜 Yunikorn?

• 기존의 k8s 스케쥴러는 CPU, Memory, GPU 등의 리소스만을 고려하여 스케쥴링을 수행하는데, 만약 다양하고 많은 양의 Pod 가 배포되고 회수되는 환경에서 필요하게되는 미세한 조정이 어려운 단점이 있다. 
• 그런 부분에서 Yunikorn 은 아래에서 소개할 다양한 기능으로 단점을 보완해준다.
• yunikorn 과 함께 많이 사용되는 volcano(2) 스케쥴러가 있다.(개인적으로 volcano 가 더 어려워보여서 유니콘을 선택했다.. 유니콘은 대시보드를 제공해주는 점이 더 매력적이었고)
• yunikorn 은 apache incubator 프로젝트고 volcano 는 cnf(cloud native computing foundation) 프로젝트이다.)

3. Features

• App-aware 스케쥴링
  
  • k8s 기본 스케쥴러와 달리 컨텍스트 정보(사용자, 애플리케이션, 큐 등) 기반의 스케쥴링을 지원한다.
  • 리소스 총량, 공정성 정책, 우선순위 정책으로 스케쥴링 fine-grained controls (적은 범위의 제어)를 할 수 있다.
• 계층적 구조의 리소스 큐
  
  • 큐 간의 계층적 구조를 통해 multi-tenancy 환경에서 효율적인 리소스 할당을 할 수 있다.
• Job Ordering and Queuing
  • 작업의 우선순위를 지정하고, 작업이 큐에 들어갈 때 우선순위에 따라 큐에 배치된다.
  • FIFO, Fair, StateAware, piriority 정책을 지원한다.
• 갱 스케쥴
  
  • 쉽게 말해 분산 환경에서 All or Nothing 스케쥴 방식이다.
  • 애플리케이션이 필요한 리소스의 세트(묶음)을 요청하고, 이 리소스가 모두 확보될 때 한 번에 스케쥴링한다.
  • 갱 스케쥴러가 placeholder pod(임시 pod)를 배정하고, UpScaling이 일어난 후 실제 pod 와 교체되는 방식이다.
  • 갱 스케쥴을 사용할 경우 FIFO 정책을 사용하게 된다. (왜냐하면 policy 는 리소스를 부분적으로 예약하기 때문에 리소스 세그먼트가 발생할 수 있다.)

4. Yunikorn 사용하기

***아래 가이드는 쿠버네티스 사용이 익숙한 사람들 기준으로 작성된 점 참고바랍니다.

A.yunikorn 설치 

2가지 설치 모드

• embeddedAdmissionController
  
  • 쿠버네티스 기본 스케쥴러 대신 유니콘 스케쥴러를 사용하는 모드
  • api server 와 통신하는 모든 트래픽을 유니콘 스케쥴러로 라우팅한다
  • 성능이 뛰어나다고 하다..ㅇㅅㅇ
• plugin Mode
  
  • 기본 스케쥴러의 framework 일부로 작동하는 모드
  •  mixed workload 에 적합하다.

그리고 helm repo values.yaml 파일 작성 (4)

service:
  type: NodePort
  targetPort: 9080
  portWeb: 9889

** 다른 설정은 하지 않았고, Dashboard ingress 생성을 위해 service 만 NodePort 로 변경했다

Ingress 생성

apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: yunikorn-ingress
  namespace: yunikorn
  annotations:
    alb.ingress.kubernetes.io/scheme: internet-facing
    alb.ingress.kubernetes.io/target-type: instance
    alb.ingress.kubernetes.io/subnets: [ 사용하는 서브넷 ]
    alb.ingress.kubernetes.io/security-groups: [보안 그룹 ]
    alb.ingress.kubernetes.io/load-balancer-name: [로드밸랜서 이름]

spec:
  ingressClassName: alb
  rules:
    - http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: yunikorn-service
                port:
                  number: 9889

helm 설치 스크립트

#!/bin/bash
helm repo add yunikorn https://apache.github.io/yunikorn-release
helm repo update


echo "install yunikorn by helm.."

helm upgrade --cleanup-on-fail \
  --install yunikorn yunikorn/yunikorn \
  --namespace yunikorn \
  --create-namespace \
  --version=1.6.0 \
  --values values.yaml

**혹시 필요한 사람을 위해 설치스크립트도 공유한다.** 

대시보드 화면

B. Yunikorn 사용

• Standard mode 로 설치했을 경우 어떤 파드로 배포하더라도 yunikorn 스케쥴러가 파드를 스케쥴링한다.
• 필자의 경우 처럼 Spark Job 을 실행할때는 아래와 같이 속성값을 추가해줘야 한다

- controller.batchScheduler.enable=true
- controller.batchScheduler.default=yunikorn

* EKS 기준으로 yunikorn 을 이용해 spark job 을 배포하는 가이드는 별도로 포스팅하고, 잘 작성된 가이드로 마무리하겠다. (3)

5.출처

•  이 포스팅에 대한 설명은 함께 작업했던 강인호(aws 코리아)님께서 공유해주신 자료를 복습하고 요약하였다.

6.References

(1) https://yunikorn.apache.org/
(2) https://volcano.sh/en/
(3) https://docs.aws.amazon.com/ko_kr/emr/latest/EMR-on-EKS-DevelopmentGuide/tutorial-yunikorn.html

(4)  https://artifacthub.io/packages/helm/yunikorn/yunikorn