technology_cloud

▶ Shard 란?

• Elasticsearch에서 index의 샤드(shard)는 데이터를 분할하여 저장하는 단위입니다. 
• 샤드는 인덱스 내의 데이터를 물리적으로 분산하여 저장하는 역할을 하며, 이는 Elasticsearch가 분산 시스템이기 때문에 여러 서버(노드)에 데이터를 고르게 분배하고 성능을 최적화하기 위해 사용됩니다.
• 샤드는 저장소로서의 역할을 하기 보다는, 인덱스 데이터를 분할하고, 데이터를 분산 저장하기 위한 개념입니다. 각 샤드는 하나의 루씬 인덱스(Lucene index)로, 독립적으로 데이터를 저장하고 검색할 수 있습니다. 그래서 각 샤드는 자체적으로 데이터를 저장하는 저장소처럼 기능하지만, 전체적으로는 Elasticsearch 클러스터 내의 여러 노드에서 데이터를 분산 처리하는 방식입니다.

▶ 샤드의 역할

1.데이터 분산 저장: 인덱스 데이터를 여러 샤드로 나누어 여러 노드에 분산시켜 저장합니다. 이는 Elasticsearch가 분산

   시스템이기 때문에 데이터를 여러 서버에 효율적으로 저장하고 처리할 수 있게 합니다.
2.성능 향상: 샤드가 여러 노드에 분배되어 저장됨으로써, 검색 요청이 여러 노드에서 병렬로 처리될 수 있습니다. 

3.확장성: 샤드를 추가하거나 기존 샤드를 다른 노드로 이동시켜서 Elasticsearch 클러스터를 확장할 수 있습니다.

프라이머리 샤드 (Primary Shard): 실제 데이터를 저장하는 기본 샤드입니다. 하나의 인덱스는 고정된 수의 프라이머리 샤드를 가집니다.
레플리카 샤드 (Replica Shard): 프라이머리 샤드의 복제본으로, 데이터의 고가용성(availability)과 내결함성(fault tolerance)을 높입니다. 레플리카 샤드는 데이터가 여러 복제본으로 저장되도록 해 주며, 검색 성능에도 도움이 됩니다.

▶ 인덱스 생성 시 샤드 수 지정

샤드 수는 인덱스를 생성할 때 지정하며, 이후에는 변경할 수 없습니다(샤드 수를 변경하려면 인덱스를 삭제하고 재생성해야 합니다). 예를 들어, 다음과 같은 명령으로 인덱스를 생성할 수 있습니다.

PUT /my_index
{
  "settings": {
    "number_of_shards": 3,     // 샤드 수
    "number_of_replicas": 2    // 레플리카 수
  }
}

-> 노드 수와 샤드 수는 독립적으로 설정되지만, 샤드를 잘 분배하고 효율적으로 활용하기 위해 노드 수와의 균형을 맞추는 것이 중요합니다.

예를 들어, 

인덱스가 3개의 primary shards와 2개의 replica shards로 구성된 경우, 이 인덱스는 총 9개의 샤드를 가집니다(3개의 primary + 6개의 replica). 만약 primary shard 중 하나가 장애로 인해 사용 불가능해지면, replica shard 중 하나가 primary shard로 승격되어 활성 상태가 됩니다. 이때, active primary shard가 새로운 복제본을 통해 활성화됩니다.
Active shard는 이처럼 primary와 replica 모두를 포함한 상태로, 검색/색인 요청을 처리할 수 있는 샤드가 어떤 노드에서든지 활성화될 수 있음을 의미합니다.


1. Active Shard (활성 샤드)
Active shard는 현재 Elasticsearch 클러스터에서 활성화되어 있는 샤드를 의미합니다. 즉, 클러스터의 어떤 노드에 할당되어 있고 데이터가 저장되고 검색 요청을 처리할 수 있는 상태인 샤드를 가리킵니다.

샤드가 활성화되었다는 것은 해당 샤드가 클러스터의 노드에 할당되어 있고 검색 및 색인 작업이 이루어질 수 있다는 의미입니다.
Active shard는 primary shard일 수도 있고, replica shard일 수도 있습니다. 즉, 주 샤드(primary)와 복제 샤드(replica) 모두가 활성 샤드로 존재할 수 있습니다.

2. Active Primary Shard (활성 기본 샤드)
Active primary shard는 특정 인덱스의 primary shard가 활성 상태로 클러스터 내의 어떤 노드에 할당되어 있다는 의미입니다.

Primary shard는 데이터를 처음 저장하는 샤드로, 인덱스 생성 시 설정한 샤드 수에 따라 기본적으로 결정됩니다.
Primary shard는 해당 데이터를 최초로 저장하고 업데이트하는 샤드로, 이후 replica shard가 primary shard의 복제본을 생성합니다.
Active primary shard는 primary shard가 현재 클러스터에서 활성화되어 데이터를 저장하거나 검색할 수 있는 상태를 의미합니다.

요약정리
Active shard는 Elasticsearch 클러스터 내에서 활성화된 샤드 전체를 의미합니다. 이 샤드는 primary 또는 replica일 수 있으며, 실제로 데이터를 처리할 수 있는 샤드를 나타냅니다.
Active primary shard는 인덱스의 primary shard 중에서 현재 활성화되어 데이터를 처리하는 샤드를 의미합니다. primary shard는 인덱스에서 원본 데이터를 저장하고 업데이트하는 역할을 하며, replica shard는 primary shard의 복제본입니다.

Airflow 란?

Apache Airflow는 데이터 파이프라인을 자동화하고, 작업을 스케줄링하며, 워크플로우를 관리하는 데 사용되는 오픈 소스 도구입니다. 머신러닝 작업에 있어 Airflow는 데이터 수집, 모델 훈련, 모델 평가, 배포 등의 다양한 단계에 걸쳐 파이프라인을 효율적으로 관리할 수 있게 도와줍니다. 또한, Airflow는 Kubernetes(K8s) 환경과의 호환성도 제공하여 분산 시스템에서 작업을 쉽게 스케줄링하고 관리할 수 있습니다.

1. Apache Airflow 개념

Apache Airflow는 기본적으로 **DAG(Directed Acyclic Graph)**를 사용하여 파이프라인을 정의하고, 이 DAG 내에서 작업(Tasks)을 정의하여 순차적, 병렬적, 조건부로 실행할 수 있게 해줍니다. DAG는 각 작업을 어떻게 실행할지에 대한 흐름과 의존성을 명시합니다.

• DAG (Directed Acyclic Graph): DAG는 작업들의 실행 순서를 정의하는 그래프입니다. 각 작업은 Task로 정의되며, Task 간 의존성은 set_upstream, set_downstream 등을 통해 설정합니다.
• Operator: Airflow는 다양한 유형의 작업을 정의하기 위한 Operator를 제공합니다. 예를 들어, PythonOperator, BashOperator, EmailOperator 등이 있습니다. 각각의 Operator는 특정 종류의 작업을 실행하는 역할을 합니다.
• Scheduler: Airflow의 스케줄러는 설정된 시간에 DAG를 주기적으로 실행합니다. 스케줄러는 DAG가 주기적으로 실행되도록 관리합니다.
• Executor: Airflow는 여러 종류의 Executor를 제공하여 작업이 어떻게 실행될지 제어합니다. SequentialExecutor, LocalExecutor, CeleryExecutor, KubernetesExecutor 등이 있습니다.

2. Airflow를 머신러닝에 활용하는 방법

Airflow는 머신러닝 프로젝트의 여러 단계를 자동화하는 데 유용하게 사용됩니다. 예를 들어:

• 데이터 파이프라인 관리: 데이터 수집, 전처리, 특성 추출 등 데이터를 준비하는 작업을 관리합니다.
• 모델 훈련 및 튜닝: 훈련 데이터를 모델에 전달하고, 모델 훈련 및 하이퍼파라미터 튜닝 작업을 자동화합니다.
• 모델 평가 및 테스트: 훈련된 모델을 테스트하고 성능을 평가하는 작업을 자동으로 수행합니다.
• 모델 배포: 훈련된 모델을 배포하여 예측 서비스를 제공하는 단계도 자동화할 수 있습니다.
• 모니터링 및 알림: 머신러닝 파이프라인의 각 단계가 성공적으로 완료되었는지 확인하고, 실패 시 알림을 제공합니다.

3. Kubernetes 환경에서의 Airflow 설치 및 사용법

Kubernetes는 Airflow와 잘 통합되며, KubernetesExecutor를 사용하여 Airflow의 작업을 Kubernetes에서 실행할 수 있습니다. 이 방법은 클러스터 리소스를 효율적으로 사용하고, 분산 환경에서 작업을 실행하는 데 유리합니다.

Kubernetes에서 Airflow 설치 방법

<1. Helm을 통한 Airflow 설치>

Helm은 Kubernetes에서 애플리케이션을 관리하고 배포하는 데 유용한 패키지 관리자입니다. Helm을 사용하여 Airflow를 설치하는 방법은 매우 간단합니다.

설치 순서:

   1-1. Helm 설치 (Helm이 설치되지 않은 경우)

# Helm 설치 (Mac 예시)
brew install helm

  1-2. Airflow Helm 차트 설치: Airflow 공식 Helm 차트를 사용하여 Kubernetes 클러스터에 Airflow를 설치합니다.

# Airflow의 Helm 차트를 설치할 수 있는 저장소 추가
helm repo add apache-airflow https://airflow.apache.org
helm repo update

  1-3. Airflow 설치: Kubernetes 클러스터에 Airflow를 설치합니다.

helm install airflow apache-airflow/airflow \
  --namespace airflow \
  --create-namespace \
  --set executor=KubernetesExecutor \
  --set airflow.image.repository=apache/airflow \
  --set airflow.image.tag=2.5.0 \
  --set airflow.image.pullPolicy=Always \
  --set redis.enabled=true \
  --set postgresql.enabled=true

위 명령은 Kubernetes 클러스터에 Airflow를 설치하고, KubernetesExecutor를 사용하여 작업을 실행하도록 설정합니다. Redis와 PostgreSQL은 기본적으로 사용되는 데이터베이스로 설정됩니다.

 1-4. Airflow 웹 UI 접근: Helm 차트를 사용하여 Airflow를 설치하면 웹 UI를 통해 Airflow 대시보드에 접근할 수 있습니다. 설   치 후,kubectl port-forward명령어로 웹 UI에 접근합니다.

kubectl port-forward svc/airflow-webserver 8080:8080 --namespace airflow

  웹 브라우저에서 http://localhost:8080으로 접속하여 Airflow UI에 로그인할 수 있습니다.

1-5. KubernetesExecutor 설정: KubernetesExecutor는 Airflow에서 Kubernetes 클러스터 내에서 작업을 실행하는 방식입니다. 각 작업이 독립적인 Pod로 실행되므로, Kubernetes의 자원을 효율적으로 활용할 수 있습니다.

  Helm 설치 시 --set executor=KubernetesExecutor를 추가하여 이 설정을 적용합니다.

<2. Airflow KubernetesExecutor 사용법>

Airflow에서 KubernetesExecutor를 사용하면 각 작업이 Kubernetes Pod로 실행됩니다. 이를 통해 작업을 확장 가능하게 관리할 수 있으며, Kubernetes의 자동 스케일링, 리소스 할당 등을 활용할 수 있습니다.

• Executor 설정: Airflow를 설치할 때 --set executor=KubernetesExecutor를 설정합니다.
• PodOperator 사용: KubernetesPodOperator는 Airflow의 작업을 Kubernetes Pod에서 실행할 수 있도록 해줍니다.

예를 들어, Python 작업을 Kubernetes Pod에서 실행하려면 KubernetesPodOperator를 다음과 같이 사용합니다:

from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from airflow import DAG
from datetime import datetime

dag = DAG('ml_workflow', start_date=datetime(2024, 1, 1))

task = KubernetesPodOperator(
    namespace='default',
    image="python:3.8",
    cmds=["python", "-c", "print('Hello, Kubernetes!')"],
    name="python-task",
    task_id="run-python-task",
    get_logs=True,
    dag=dag,
)

이 예시에서 KubernetesPodOperator는 Python 3.8 이미지를 사용하여 Kubernetes에서 Python 작업을 실행합니다.

Kubernetes에서 리소스 설정: 작업이 실행될 때마다 리소스(CPU, 메모리) 요구 사항을 설정할 수 있습니다. 예를 들어, 작업에 대해 최소한의 CPU나 메모리 리소스를 할당할 수 있습니다.

task = KubernetesPodOperator(
    namespace='default',
    image="python:3.8",
    cmds=["python", "-c", "print('Hello, Kubernetes!')"],
    name="python-task",
    task_id="run-python-task",
    get_logs=True,
    resources={'request_cpu': '500m', 'request_memory': '1Gi'},
    dag=dag,
)

4. 기타 고려사항

• 헬름 차트 설정: Helm 차트에 포함된 다양한 설정을 통해 Airflow 클러스터의 리소스를 설정하고, 네트워크 및 데이터베이스 연결 등을 맞출 수 있습니다.
• 모니터링: Kubernetes와 결합한 Airflow는 Prometheus와 Grafana 같은 모니터링 도구와 쉽게 통합할 수 있습니다. 이를 통해 파이프라인의 상태를 실시간으로 확인할 수 있습니다.

AirFlow 도입 사례:

https://www.bucketplace.com/post/2021-04-13-%EB%B2%84%ED%82%B7%ED%94%8C%EB%A0%88%EC%9D%B4%EC%8A%A4-airflow-%EB%8F%84%EC%9E%85%EA%B8%B0/