데이터 엔지니어링

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

오늘은 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

1.Dag Factory 란?

• yaml 포맷 기반의 워크플로우 정의를 읽어 dag 을 자동으로 생성해주는 에어플로우 라이브러리이다. (1)(2)

2. 왜 Dag Factory ? 

• Airflow 를 사용하려면 아래와 같은 절차로 Workflow 를 만드는 작업을 해야 하는데, 이런 수고를 줄여준다.
• workflow 는 dag 객체를 생성하는 파이썬 스크립트 파일을 만들어야 하기 때문에 파이썬 문법을 알아야 한다.
• dag 파일을 작성하려면 파이썬의 문법과 airflow 에서 제공하는 core 패키지 문서를 학습하여 작성해야 한다.
• 워크플로우들이 많아질때는 관리하기 힘들어지기 때문에, Dag의 메타데이터화와 동시에 자동화를 고려해야 하는 시기가 온다. 
  
  • python 과 airflow 의 주요 내용들의 학습 없이 DAG 구조체를 만들어준다
  • 중복된 코드를 피해준다

3. Features

• Multiple Configuration Files
• Custom Operators
• Callbacks 지원

4. Dag Factory 사용하기

절차

1) dag factory 를 설치한다.
2) 쿠버네티스를 사용할 경우 2가지 방법이 있다.
           a.패키지 설치 버전의 이미지를 직접 빌드하는 방법
           b.pod 내 airflow 가 init 할때 패키지 설치 명령어를 추가하는 방법
3) yaml 포맷의 워크플로우 파일을 작성한다.
4) dagfactory 객체를 이용해 yaml 리소스를 로드할 python 파일을 작성한다.
5) 작성한 2개의 파일을 dag 폴더로 이동시킨다

A. 설치

$ pip install dag-factory

B.쿠버네티스에서 helm 을 이용해 배포할 경우

쿠버네티스에서 사용할 경우 아래 두 가지 선택이 있을 수 있다.
첫 번째 ) 직접 이미지를 빌드하는 방법
두 번째 ) pod 내 airflow 가  init 할때 패키지 설치 명령어를 추가하는 방법

1)  dag-factory 를 설치완료한 이미지를 빌드하여 올려서 사용한다.

Docker file 예시 

FROM apache/airflow:2.8.2

RUN pip install dag-factory==0.19.0

ENTRYPOINT ["/usr/bin/dumb-init", "--", "/entrypoint"]
CMD []

 2) airflow chart values 파일 내 worker 가 bootstrap 할 때 패키지 설치 명령어를 추가한다.

workers:
  # Number of airflow celery workers in StatefulSet
  replicas: 1
  # Max number of old replicasets to retain
  revisionHistoryLimit: ~

  # Command to use when running Airflow workers (templated).
  command: ~
  # Args to use when running Airflow workers (templated).
  args:
    - "bash"
    - "-c"
    # The format below is necessary to get `helm lint` happy
    - |-
      python -m pip install --upgrade pip & pip install --no-cache-dir dag-factory && exec \
      airflow {{ semverCompare ">=2.0.0" .Values.airflowVersion | ternary "celery worker" "worker" }}

C.사용

절차
1 ) yaml 포맷의 워크플로우 파일 작성한다.
2) dagfactory 객체를 이용해 yaml 리소스를 로드할 python 파일을 작성한다.
3) 작성한 2개의 파일을 dag 폴더로 이동시킨다

1) yaml 포맷의 워크플로우 파일 작성하기

default:
  default_args:
    catchup: false,
    start_date: 2024-11-11

basic_example_dag:
  default_args:
    owner: "custom_owner"
  description: "this is an example dag"
  schedule_interval: "0 3 * * *"
  render_template_as_native_obj: True
  tasks:
    task_1:
      operator: airflow.operators.bash_operator.BashOperator
      bash_command: "echo 1"
    task_2:
      operator: airflow.operators.bash_operator.BashOperator
      bash_command: "echo 2"
      dependencies: [task_1]
    task_3:
      operator: airflow.operators.bash_operator.BashOperator
      bash_command: "echo 2"
      dependencies: [task_1]

2) python 파일 작성하기

import os
from pathlib import Path

# The following import is here so Airflow parses this file
# from airflow import DAG
import dagfactory

DEFAULT_CONFIG_ROOT_DIR = "/usr/local/airflow/dags/"

CONFIG_ROOT_DIR = Path(os.getenv("CONFIG_ROOT_DIR", DEFAULT_CONFIG_ROOT_DIR))

config_file = str(CONFIG_ROOT_DIR / "example_dag_factory.yml")

example_dag_factory = dagfactory.DagFactory(config_file)

# Creating task dependencies
example_dag_factory.clean_dags(globals())
example_dag_factory.generate_dags(globals())

3) 파일을 dag 디렉토리로 이동시킨다.

mv * /usr/local/airflow/dags/

** 이전에 포스팅한 GitSync 를 같이 이용할 수 있습니다. 그렇게 되면 dag 파일을 디렉토리로 이동할 필요 없이 git repository 로 이동하면 되는거죠. :) [3]

화면 참고용 스크린샷

5.References

[1]  https://github.com/astronomer/dag-factory
[2]  https://astronomer.github.io/dag-factory/latest/getting-started/quick-start-airflow-standalone/

[3] https://jssvs.tistory.com/101