파이프라인 구축 (with NES / notebook exceute system)

 

해당 글은 아래의 시스템 구조를 참고하여 만든 예제 입니다.

https://devocean.sk.com/vlog/seminar/20220728_Data_Engineering_in_SKT.pdf

https://netflixtechblog.com/scheduling-notebooks-348e6c14cfd6

Part 2: Scheduling Notebooks at Netflix

 

완성 코드  (gihtub 링크)

https://github.com/uiandwe/nes/tree/main

GitHub - uiandwe/nes

 

 

 

 

1. 어떻게 AI팀에서 만든 코드를 바로 적용할수 있을까? (with mlops)

앞선 글 mlops 아키텍처 [https://uiandwe.tistory.com/1392] 에서 ML / ops가 단절된 상태를 이야기 한부분이 있습니다.

이런 부분이 발생하면 코드의 정합성과 구현에 대해서 정확한 값을 산출할수 없고, ML 코드를 다시 ops에 맞게 컨버팅하는 시간이 들어가게 됩니다. (일을 두번하게 됩니다)

 

대부분의 ML의 목적은 인퍼런스의 값을 통해 사용자에게 전달하고 사용자 피드백 반영을 통한 챗봇의 동적인 학습이 필요합니다.

이 과정을 도와주는게 mlops입니다. 그러면 어떻게 ML의 코드를 바로 ops에서 사용하도록 할수 있을까요?

 

먼저 데이터를 수집 및 가공 / ML모델을 주기적으로 생성하기 위해서 airflow를 사용합니다.

airflow로 개발을 하게 된다면 dag파일을 코딩하고 airflow에 배포 및 실행하는 과정으로 개발하게 되겠죠.

문제는 해당 과정이 디버깅이 불편하며 (오직 print로 확인가능), 개발 환경이나 생산성이 떨어집니다.

특히나 AI나 데이터사이언스에서 넘겨지는 대부분이 jupyter notebook을 이용해서 개발을 하게 되는데,

해당 notebook을 데이터파이프라인에 사용하기 위해 dag.py로 포팅하는 작업이 발생하게 됩니다.

(개인적으로 이 포팅 과정에서 정말 많이 싸웠습니다.... 서로 다른 패키지 버전 / 알수 없는 파라미터 / 재현불가능한 로직 등등..)

 

 

ML에서 작성한 노트북을 바로 사용가능하다면!!! 생산성이 극대화 되겠죠!!!

 

2. papermill 패키지를 이용한 notebook 실행

papermill은 주피터 노트북의 매개변수화, 실행, 분석을 위한 도구입니다. 노트북 실행시 파라미터를 사용자의 작업 필요없이 매개변수를 통해 자동으로 실행가능하게 만드는 패키지 입니다. 해당 패키지를 통해 데이터 파이프라인에서의 notebook 실행을 더욱 자동화 할 수 있습니다.

 

 

 

3. Notebook Execution Service

papermill 패키지를 통해 공유된 notebook의 실험 테스트가 가능해졌습니다. 이를 airflow와 결합하게 된다면, 각 팀에서 사용한 notebook의 공유를 통해  하나의 시스템에서 재현가능하게 됩니다.

 

 

해당 글은 로컬에서 실행가능한 모듈로 만들었습니다. 특히 airflow는 local로 실행한 글입니다. 상용이나 개발에 사용시에는 airflow와 jupyter의 버전이나 혹은 필요하신 다른 시스템을 덧붙여서 사용하시기 바랍니다.

 

 

해당 글은 airflow와 jupyter의 기본 사용을 안다는 전제하에 글을 썼습니다. 사용 방법은 공식문서를 참조해주시기 바랍니다.

 

 

 

구현 실습

 

1. docker-compose airflow + jupyter 기동과 papermill operater 설치

airflow에서 사용할 papermill[https://airflow.apache.org/docs/apache-airflow-providers-papermill/stable/operators.html]

입니다. 해당 패키지를 requirements.txt에 넣어줍니다.

 

requirements.txt

requests
pandas
numpy
pathlib
beautifulsoup4
psycopg2-binary
sqlalchemy
pymongo
apache-airflow
apache-airflow-providers-ssh
apache-airflow-providers-papermill

 

airflow의 Dockerfile를 생성해 줍니다. 최신 버전인 2.6.1을 사용할 예정입니다.

FROM apache/airflow:2.6.1
USER root
RUN apt-get update \
  && apt-get install -y --no-install-recommends \
        vim \
  && apt-get autoremove -yqq --purge \
  && apt-get clean \
  && rm -rf /var/lib/apt/lists/*
USER "${AIRFLOW_UID:-50000}:0"





# requirements.txt 복사
COPY requirements.txt .



# 필요한 Python 패키지 설치
RUN pip install --upgrade pip
RUN pip install --no-cache-dir -r requirements.txt

 

 

 

docker-compose.yml의 파일은 다음과 같습니다.

airflow 실행과 함께 유저를 생성하며(admin/ admin) jupyter와 같은 폴더를 공유하는것을 볼수 있습니다.

version: '3'
services:
  airflow:
    build: .
    command: >
      bash -c '
      airflow db init &&
      airflow users create --username admin --password admin --firstname Anonymous --lastname Admin --role Admin --email test@test.com &&
      (airflow webserver & airflow scheduler & wait)'
    ports:
      - 8080:8080
    volumes:
      - ./dags:/opt/airflow/dags
      - ./data:/opt/airflow/data
      - ./output:/opt/airflow/output
  jupyter:
    image: 'jupyter/base-notebook:latest'
    volumes:
      - ./dags:/opt/airflow/dags
      - ./data:/opt/airflow/data
      - ./output:/opt/airflow/output
    ports:
        - "8888:8888"
    command: "start-notebook.sh \
              --NotebookApp.password='' \
              --sudo mkdir /opt/airflow/dags"

 

이제 docker-compose up으로 실행하면, airflow를 docker 이미지를 생성한 후 기동하게 됩니다.

localhost:8080으로 접속한다면 airflow를 로그인할수 있습니다. (admin // admin)

메뉴의 providers를 들어가게 되면 papermill이 정상적으로 설치 되었음을 확인할수 있습니다.

 

 

 

 

이번엔 jupyter로 접속합니다. jupyter는 localhost:8888 입니다.

해당 폴더에 아래와 같은 dag파일을 하나 생성했습니다.

papermill를 사용해서 notebook를 실행하는 코드 입니다.

파라미터는 notebook이 있는 경로 / 실행후 로그파일 경로 / notebook에 입력될 파라미터 입니다.

from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.operators.bash import BashOperator
from airflow.providers.papermill.operators.papermill import PapermillOperator
from airflow import DAG
from datetime import datetime, timedelta
import airflow.utils.dates
import json
import os



def print_context(ds=None, **kwargs):
    """Print the Airflow context and ds variable from the context."""
    print(kwargs)
    print(ds)
    return "Whatever you return gets printed in the logs"


with DAG(
    dag_id="example_papermill_operator_verify",
    schedule='*/10 * * * *',
    start_date=datetime(2023,12,7),
    catchup=False,
) as dag:
    
    run_bash = BashOperator(
        task_id="run_bash",
        bash_command="echo start dag",
    )
    
    run_example_notebook = PapermillOperator(
        task_id="run_example_notebook",
        input_nb=os.path.join("/opt/airflow/dags/notebook/test_papermill.ipynb"),
        output_nb="/opt/airflow/output/out-{{ execution_date }}.ipynb",
        parameters={
                    "msgs": "Ran from Airflow at {{ execution_date }}!",
                    "execution_date": "{{execution_date}}" 
                   },
    )
    
    done_task = PythonOperator(task_id="done_task", python_callable=print_context)

    run_bash >> run_example_notebook >> done_task

 

 

해당 경로에 dags/notebook/test_papermill.ipynb도 추가해줍니다.

파라미터로 받은 데이터를 출력하는 간단한 notebook입니다. (파라미터가 잘 들어갔는지 확인용)

test_papermill.ipynb

0.00MB

 

 

 

2. dag 실행 및 결과 확인

dag를 실행하게 되면 아래와 같은 에러가 발생할겁니다. 바로 notebook 실행시의 python 커널을 찾지 못한 에러 입니다.

ERROR: jupyter_client.kernelspec.NoSuchKernel: No such kernel named python3 occurs in Airflow/papermill running as a docker container

 

https://stackoverflow.com/questions/69759351/error-jupyter-client-kernelspec-nosuchkernel-no-such-kernel-named-python3-occu

ERROR: jupyter_client.kernelspec.NoSuchKernel: No such kernel named python3 occurs in Airflow/papermill running as a docker cont

이는 airflow 생성시 해당 커널을 생성해 주면 됩니다. Dockerfile에 사용할 커널 정보를 추가해줘야 합니다.

# pamermill 에 쓰일 커널 설정
RUN pip install --upgrade pip ipython ipykernel
RUN ipython kernel install --name "python3" --user

 

 

 

최종 Dockerfile의 코드는 다음과 같습니다.

(만일 jupyter kernel에서 필요한 패키지가 있다면 사용 커널 생성시 패키지를 설치해 주면 됩니다)

FROM apache/airflow:2.6.1
USER root
RUN apt-get update \
  && apt-get install -y --no-install-recommends \
        vim \
  && apt-get autoremove -yqq --purge \
  && apt-get clean \
  && rm -rf /var/lib/apt/lists/*
USER "${AIRFLOW_UID:-50000}:0"



# pamermill 에 쓰일 커널 설정
RUN pip install --upgrade pip ipython ipykernel
RUN ipython kernel install --name "python3" --user


# requirements.txt 복사
COPY requirements.txt .



# 필요한 Python 패키지 설치
RUN pip install --upgrade pip
RUN pip install --no-cache-dir -r requirements.txt

 

 

다시 docker를 빌드 한 후 docker-compose를 실행하고, dag을 실행하게 되면 로그를 남기려 했던 경로에 out-***으로 시작하는 ipynb가 생성된것을 확인할수있습니다.

 

보통은 현업에서는 해당 실행 로그를 s3에 올리게 됩니다. 실행 이력과 함께 공유를 위해서 스토리제에 적재하는 형식으로 구성합니다.

 

 

 

3. ipynb 파일 뷰어 설치

콘솔에서 실행 완료의 파일을 볼수있는건 귀찮으므로, 바로 볼수 있는 뷰어를 추가해줍니다. 저는 jnotebook_reader라는 프로그램을 추가할 예정입니다.

 

https://github.com/line/jnotebook_reader

GitHub - line/jnotebook_reader: :black_joker: An awesome viewer to browse and render Jupyter Notebooks from local, Amazon S3, Go

 

해당 git 소스를 받은 후 도커 이미지를 생성합니다.  테스트로 실행을 해보면 정상적으로 서버가 올라온것을 볼수 있습니다. (기본으로 제공되는 파일들이 보입니다.)

docker build -t jnotebook_reader -f docker/Dockerfile .
docker run -p 9088:9088 \
    -e JNOTEBOOK_READER_SERVER_PORT="9088" \
    -it --rm jnotebook_reader

 

 

 

 

 

 

이제 docker-compose에 적용해 줍니다. docker-compose.yml에 jnotebook-reader를 추가해줍니다.

로그를 같이 공유하기 위해 로그가 쌓이는 폴더를 볼륨에 추가해줍니다.

최종 docker-compose.yml은 다음과 같습니다.

 

version: '3'
services:
  airflow:
    build: .
    command: >
      bash -c '
      airflow db init &&
      airflow users create --username admin --password admin --firstname Anonymous --lastname Admin --role Admin --email test@test.com &&
      (airflow webserver & airflow scheduler & wait)'
    ports:
      - 8080:8080
    volumes:
      - ./dags:/opt/airflow/dags/
      - ./data:/opt/airflow/data
      - ./output:/opt/airflow/output
      - ./secrets.json:/etc/airflow/secrets.json
  jupyter:
    image: 'jupyter/base-notebook:latest'
    volumes:
      - ./dags:/opt/airflow/dags/
      - ./data:/opt/airflow/data
      - ./output:/opt/airflow/output
    ports:
        - "8888:8888"
    command: "start-notebook.sh \
              --NotebookApp.password='' \
              --sudo mkdir /opt/airflow/dags"

  commuter:
    image: jnotebook_reader:latest
    ports:
      - "9088:9088"
    volumes:
      - ./data:/opt/airflow/data
      - ./output:/opt/airflow/output

 

 

다시 dag을 실행 한 후 localhost:9088로 접속해보면 실행된 로그들을 볼수 있습니다. 클릭해서 보시면

아래와 같이 dag에서 설정한 parameters가 정상적으로 찍힌것을 볼수 있습니다.

 

 

 

next level

해당 글의 실습에서는 CI/CD가 빠져 있습니다. 예를 들어 jupyter notebook을 github에 push하면 자동으로 코드가 반영되어, 다음 dag실행시에는 변경된 notebook이 실행되도록 하는 로직이 필요합니다.

이는 github의 action 이나 CI를 제공하는 서비스를 통해 구현이 가능합니다. (ex : push 트리거를 통해 NES가 있는 서버에 배포)

CI/CD까지 완성한다면 더욱 완벽한 데이터파이프라인이 완성될 겁니다.