이번 글에서는 이러한 오류가 발생한 원인과, 이를 해결하기 위한 설정 방법에 대해 정리해보겠습니다.

🔄 gitSync의 .worktree 구조 이해

gitSync는 내부적으로 .worktree 디렉토리를 생성하고, 여기에 커밋 해시를 이름으로 하는 서브 디렉토리를 만들어 소스를 싱크합니다. 예를 들어 다음과 같이 구조화됩니다.

.worktree/
  ├── abc123456789abcdef/  ← 특정 커밋의 소스
  └── def987654321fedcba/  ← 최신 커밋의 소스

이 .worktree 내부 디렉토리 중 가장 최신 커밋 폴더가 repo라는 심볼릭 링크로 연결되어 실제 실행 시 사용됩니다.

⛔️ Task는 옛날 소스를 사용 중인데...

기본적으로 gitSync는 1개의 커밋만 보관합니다. 즉, 새 커밋이 동기화되면 이전 커밋 디렉토리는 바로 삭제됩니다.

문제는, Airflow task가 수행되는 중 sync가 발생하면 이전 커밋 디렉토리는 삭제되고 task는 이미 삭제된 경로를 참조하고 있어 file not found 에러가 발생할 여지가 있습니다.

이런 현상은 특히 실행 중인 Task가 오래 걸리는 경우에 잘 나타납니다.

🛠 해결 방법: GITSYNC_STALE_WORKTREE_TIMEOUT

gitSync는 오래된 .worktree 디렉토리를 일정 시간 보관해둘 수 있도록 GITSYNC_STALE_WORKTREE_TIMEOUT 환경변수를 제공합니다.

gitSync:
    enabled: true
env:
  - name: GITSYNC_STALE_WORKTREE_TIMEOUT
    value: "60m"  # 1시간 보관

• 세부설명은 https://github.com/kubernetes/git-sync 참고

이 값을 설정하면, gitSync는 sync가 되더라도 이전 커밋 디렉토리를 설정 시간만큼 유지합니다. 이 시간 내에 실행된 task들은 문제 없이 이전 경로를 사용할 수 있게 되는 것이죠.

해당 설정을 적용한 이후로는 file not found 오류가 더 이상 발생하지 않았습니다.

장기 실행되는 Task가 있는 환경이라면, GITSYNC_STALE_WORKTREE_TIMEOUT 설정을 고려해보시길 추천드립니다.

[서버 설정]

패키지 설치

$ sudo apt update
$ sudo apt install nfs-kernel-server

파일 시스템 생성

1. 파일 시스템 생성

   $ mkdir -p ~/srv/nfs4
   $ mkdir -p ~/srv/nfs4/data
   $ chmod 777 ~/srv/nfs4/data
   $ mkdir -p ~/data

2. 디렉토리 바인드&마운트

• NFSv4 서버를 구성할 때는 Global NFS root 디렉토리를 사용하고, 실제 디렉토리를 공유 마운트 지점에 바인딩하는 것을 권장합니다.

  $ sudo mount --bind ~/data ~/srv/nfs4/data

3. /etc/fstab 에 영구 등록

   /home/dw/data /home/dw/srv/nfs4/dw none bind 0 0

NFS 설정

1. /etc/exports 에 설정 등록

   /home/dw/srv/nfs4       192.X.X.0/24(rw,sync,no_subtree_check,crossmnt,fsid=0)
   /home/dw/srv/nfs4/data  192.X.X.X(rw,sync,no_subtree_check) 192.X.X.X(rw,sync,no_subtree_check)

• 서브넷 클라이언트에만 NFS 볼륨에 대한 액세스 허용하며 하위 디렉토리 공유
• 읽기 및 쓰기 액세스는 특정 IP만 허용

2. 설정 반영

   $ sudo exportfs -ra

3. 설정 확인

   $ sudo exportfs -v 

방화벽 작업

sudo ufw allow from 192.X.X.0/24 to any port nfs

[클라이언트 설정]

패키지 설치

$ sudo apt update
$ sudo apt install nfs-common

NFS 폴더 마운트

$ sudo mount -t nfs -o vers=4 192.168.10.203:/data /data
# 마운트 해제
$ sudo umount -f /data

• 앞에서 지정한 NFS 옵션 fsid=0 으로 인해 NFS root 폴더가 /로 인식되므로 해당 사항을 고려해 마운트 폴더를 지정해야 NFS v4로 설정됩니다.

마운트 확인

$ df -h 

어쩌다 StatsD exporter -> Prometheus 루트가 아니라 Telegraf -> InfluxDB 루트를 타게된 썰😅

• 보통 Airflow metric 적재 시에는 statsd_exporter -> Prometheus -> Grafana 아키텍처를 사용합니다만 모종의 이유로 InfluxDB를 사용해야했고, Telegraf -> InfulxDB 아키텍처로 metric을 수집한 기록을 정리하고자 합니다.
• Airflow와 Telegraf는 helm을 이용해서 구축했습니다.

StatsD

• StatsD는 애플리케이션에서 발생하는 성능 지표를 수집하고, 이를 모니터링 시스템에 전송하는 데 사용되는 프로토콜입니다.
• airflow에서 생성된 메트릭은 StatsD를 통해서 전송됩니다.
• StatsD를 사용하기 위해서 helm airflow에서 아래와 같이 설정해줍니다.

values.yaml

statsd:
  enabled: true
  ...

config:
  metrics:
    statsd_on: '{{ ternary "True" "False" .Values.statsd.enabled }}'
    statsd_port: 9125           # StatsD 서버와 통신할 포트
    statsd_prefix: 'airflow'    # StatsD로 전송되는 메트릭의 접두사
    statsd_host: 'telegraf'     # StatsD 데이터를 수신할 서버의 호스트 이름

airflow metric

• 위와 같이 설정하고 나면 tcpdump -i any udp port 9125 -A 등의 네트워크 조회 명령어를 통해 실제로 어떤 메트릭을 내보내고 확인할 수 있습니다.
• 제공하는 전체 메트릭은 해당 문서에서도 자세히 제공하고 있습니다.

Telegraf

• Telegraf는 여러 input plugin을 지원하고 있고, 여기서는 Airflow와 k8s 클러스터의 메트릭을 모두 수집하기 위해 stasd, kubernetes plugins을 사용합니다.
• 메트릭 저장소는 influxDB를 사용할 것이기 때문에 outputs은 influxDB로 지정해줍니다.
• StatsD exporter를 사용할 경우, statsd-exporter mapping 파일을 통해 메트릭 변환 방식을 정의하지만 여기서는 StatsD exporter를 사용하지 않고 Telegraf를 사용하기 때문에 Telegraf config에서 템플릿을 지정합니다.
• kubernetes input plugin은 Kubelet API를 통해 메트릭을 수신하는 메커니즘입니다. (kubernetes 메트릭)
  • 단일 호스트에 대한 수집을 지원함으로 클러스터 내, 모든 노드에 배포가 필요합니다.

values.yaml

config:
  ...
  outputs:
    - influxdb_v2:
        urls:
          - "${urls}"
        organization: "${organization}"
        bucket: "${bucket}"
        token: "${token}"
   inputs:
    - statsd:
        service_address: ":9125"
        metric_separator: "_"
        data_format: "influx"
        templates:
          - "*.dag.*.*.duration measurement.measurement.dag_id.task_id.measurement"
          - "*.dagrun.dependency-check.* measurement.measurement.measurement.dag_id"
          - "*.dag_processing.last_duration.* measurement.measurement.measurement.dag_id"
          - "*.dagrun.duration.*.* measurement.measurement.measurement.status.dag_id"
          - "*.dagrun.schedule_delay.* measurement.measurement.measurement.dag_id"
          - "*.dag_processing.last_runtime.* measurement.measurement.measurement.dag_id"
          - "*.dag.loading-duration.* measurement.measurement.measurement.dag_id"
          - "*.operator.*.* measurement.measurement.status.operator"
    - kubernetes:
        url: "https://$K8S_NODE_IP:10250"
        bearer_token: "/var/run/secrets/kubernetes.io/serviceaccount/token"
        tls_ca: "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
        insecure_skip_verify: true

InfluxDB

• 위와 같이 설정을 하게되면 InfluxDB에서 Airflow와 k8s 두 메트릭을 모두 확인할 수 있고, Grafana에서 해당 메트릭을 시각화할 수 있습니다.

• DAG 실행시간 모니터용 operator
• 현재 DAG의 수행시간이 평균 DAG 수행시간보다 2배 이상 걸릴 경우 slack에 alert을 합니다.
  • 히스토리 DAG는 success 상태를 기준으로 합니다.

사용방법

• CheckRunTimeOperator 를 DAG 파일에 import 후 가장 마지막 task로 지정합니다.
• parameter
  • num_dags : 평균 실행시간 계산의 기준이 되는 DAG의 개수(최신순)

소스

• CheckRunTimeOperator

  from datetime import timedelta, datetime, timezone
  

import common.constant.Global from airflow.models import BaseOperator from airflow.models import DagRun

from airflow import settings

class CheckRunTimeOperator(BaseOperator):

def __init__(
    self,
    num_dags: int = 3,
    *args, **kwargs
):
    super().__init__(*args, **kwargs)
    self.num_dags = num_dags

def execute(self, context):
    dag_id = context['dag_run'].dag_id
    num_dags = self.num_dags
    session = settings.Session()
    last_runs = session.query(DagRun).filter(
        DagRun.dag_id == dag_id,
        DagRun.state == 'success'
    ).order_by(DagRun.execution_date.desc()).limit(num_dags).all()

    now_run = session.query(DagRun).filter(
        DagRun.dag_id == dag_id,
        DagRun.state == 'running'
    ).order_by(DagRun.execution_date.desc()).limit(1).all()

    if len(last_runs) < num_dags:
        return

    start_date = datetime.fromisoformat(str(now_run[0].start_date))
    duration = (datetime.now(timezone.utc) - start_date).seconds

    total_run_time = sum([run.end_date - run.start_date for run in last_runs], timedelta(0))
    avg_run_time = (total_run_time / num_dags).seconds

    if duration > avg_run_time * 2:
        context['avg_run_time'] = f'{avg_run_time // 60}m {avg_run_time % 60}s'
        context['duration'] = f'{duration // 60}m {duration % 60}s'
        # 슬랙 alert

- TEST DAG
```python
import os
from datetime import datetime

import pendulum
from airflow.decorators import dag
from airflow.operators.python import PythonOperator
from custom.dag.check_run_time import CheckRunTimeOperator

local_tz = pendulum.timezone('Asia/Seoul')

home_path = os.path.expanduser('~')

args = {
    'owner': 'owner',
    'depends_on_past': False,
    'start_date': datetime(2023, 11, 7, tzinfo=local_tz)
}


@dag(
    default_args=args,
    description='DAG의 실행시간 모니터 예시',
    schedule_interval='0 1 * * *',
    catchup=False,
    doc_md="""
        # **example_check_run_time**
        DAG의 실행시간 모니터 operator의 사용 예시입니다.
        * CheckRunTimeOperator
            * 최근 DAG(num_dags수) 평균 실행시간 보다 현재 DAG의 실행시간이 2배 이상 소요될 경우, slack에 warning alert을 합니다.
            * parameter
                * `task_id`(str) : task id 
                * `num_dags`(int) : 평균 실행시간을 계산할 최근 DAG의 개수(defult : 3)
    """
)
def example_check_run_time():

    def test_func():
        print('test func')

    test_operator = PythonOperator(
        task_id='test_operator',
        python_callable=test_func
    )

    check_time_operator = CheckRunTimeOperator(
        task_id='check_time_operator',
        num_dags=4
    )

    test_operator >> check_time_operator


dag = example_check_run_time()

PyCharm이 알려줄 때 말을 잘 듣도록 하자...

"Python Enhancement Proposal 8"

• PEP는 파이썬 개선 제안의 약자로 파이썬이나 그 프로세스 또는 환경에 대한 새로운 기능이나 규칙을 제안하는 문서입니다.
• 그중 PEP 8은 파이썬 프로그래밍 언어의 코딩 스타일 가이드라인을 정의한 문서로 코드의 가독성을 향상시키기 위해 따라야 할 규칙과 권장 사항을 제시하고 있습니다.
• 가독성과 일관성을 위해 지켜져야하는 규칙

Code Lay-out

Indentation

• 들여쓰기에 4개의 스페이스를 사용하여햐 합니다.
• 들여쓰기는 탭이 아닌 스페이스를 기본 원칙으로한다.
• 괄호 및 괄호 안의 괄호와 같이 연결되는 라인에서 줄바꿈이 일어나는 요소들은 수직으로 정렬되어야 합니다.

Good

# 괄호로 정렬되는 경우
foo = long_function_name(var_one, var_two,
                         var_three, var_four)

# 다른 요소와 구분을 위해 더 많은 들여쓰기를 한 경우
def long_function_name(
        var_one, var_two, var_three,
        var_four):
    print(var_one)

# 매달린 형태의 들여쓰기는 하나의 들여쓰기 레벨을 추가해야함
foo = long_function_name(
    var_one, var_two,
    var_three, var_four)

Bad

# 수직정렬이 되지 않았을 경우, 첫번째 줄에는 인수가 없어야함
foo = long_function_name(var_one, var_two,
    var_three, var_four)

# 다른 요소와 구분이 되지 않는 경우
def long_function_name(
    var_one, var_two, var_three,
    var_four):
    print(var_one)

Indentation - if / 닫는 괄호

• if문이 길 경우는 if문 내의 코드와 충돌할 가능성이 있습니다. PEP 8에서는 이에 대해 몇가지 베스트 옵션을 제공합니다.
• 닫는 괄호의 경우는 마지막 줄의 첫번째 요소 혹은 공백 없이 라인의 가장 앞에 위치합니다.

Good - if

# 들여쓰기 없는 경우
if (this_is_one_thing and
    that_is_another_thing):
    do_something()

# if문 내 코드와 구분하기 위한 주석 추가
if (this_is_one_thing and
    that_is_another_thing):
    # Since both conditions are true, we can frobnicate.
    do_something()

# 추가적인 들여쓰기
if (this_is_one_thing
        and that_is_another_thing):
    do_something()

Good - 괄호

# 마지막 줄의 첫번째 요소에 위치
my_list = [
    1, 2, 3,
    4, 5, 6,
    ]
result = some_function_that_takes_arguments(
    'a', 'b', 'c',
    'd', 'e', 'f',
    )

# 라인의 가장 앞에 위치
my_list = [
    1, 2, 3,
    4, 5, 6,
]
result = some_function_that_takes_arguments(
    'a', 'b', 'c',
    'd', 'e', 'f',
)

Maximum Line Length

• 한 줄의 최대 길이를 79자로 제한합니다. (상황에 따라 99자까지 허용하기도…)
• docstrings나 comments 처럼 긴 텍스트는 한 줄의 최대 길이를 72자로 제한합니다.
• 백슬래시를 활용해 줄바꿈을 합니다.

Should a Line Break Before or After a Binary Operator?

• 이전에는 이항연산자 후에 줄바꿈을 하는 방식이 추천되었으나, 수학자들의 전통을 따라 이항연산자 전에 줄바꿈을 하는 것을 추천합니다.

Good

income = (gross_wages
          + taxable_interest
          + (dividends - qualified_dividends)
          - ira_deduction
          - student_loan_interest)

Blank Lines

• 클래스 정의와 함수 정의 사이에는 두 개의 공백 라인을 사용하여 구분하고 있습니다.
• 클래스 내부의 메서드 정의는 단일한 공백 라인으로 구분하고 있습니다.
• 관련 함수 그룹을 구분하기 위해 빈 줄을 추가로 사용할 수 있습니다

class Calculator:

    def __init__(self) -> None:
        self.value: int = 0

    def add(self, x: int, y: int) -> int:
        return x + y

    def subtract(self, x: int, y: int) -> int:
        return x - y


def main() -> None:
    calc = Calculator()

    result1 = calc.add(5, 3)
    print("Addition result:", result1)

    result2 = calc.subtract(10, 4)
    print("Subtraction result:", result2)


if __name__ == "__main__":
    main()

import

• import는 행으로 구분되어 사용되어야 합니다.
• import는 아래와 같이 그룹화 하도록 합니다.
  • 표준 라이브러리 import
  • 관련된 서브파티 import
  • 로컬 어플리케이션 / 자체 라이브러리 import
  • ** 위 그룹 사이에는 빈 줄을 넣는 것이 좋다
• 원칙적으로 Absolute import가 권장되며 wildcard import는 지양해야 합니다.

Module Level Dunder(Double under) Names

• Dunder 모듈의 경우는 닥스트링 뒤, from future import barry_as_FLUFL 을 제외하고 import문 전에 쓰여져야 합니다.

"""This is the example module.
This module does stuff.
"""

from __future__ import barry_as_FLUFL

__all__ = ['a', 'b', 'c']
__version__ = '0.1'
__author__ = 'Cardinal Biggles'

import os
import sys

Whitespace in Expressions and Statements

• 아래 같은 상황에서 관계없는 공백은 피하도록 합니다.
• 후행공백은 지양합니다.

spam( ham[ 1 ], { eggs: 2 } ) 
  => spam(ham[1], {eggs: 2})
bar = (0, ) 
  => foo = (0,)
if x == 4 : print(x , y) ; x , y = y , x
  => if x == 4: print(x, y); x, y = y, x

• 이진연산자 주변은 공백을 추가하도록 합니다.

• 다른 우선순위의 연산자가 함께 쓰인다면, 가장 우선순위가 낮은 연산자 주변에 공백을 추가하는 것을 고려합니다.

  i=i+1
  => i = i + 1
  submitted +=1
  => submitted += 1
  x = x * 2 - 1
  => x = x*2 - 1
  hypot2 = x * x + y * y
  => hypot2 = x*x + y*y
  c = (a + b) * (a - b)
  => c = (a+b) * (a-b)

• 키워드 독립변수 혹은 기본 파라미터 값을 나타내기 위해 = 주변에 스페이스를 사용을 지양합니다.

• type hint와 기본 파라미터 값을 같이 쓸 경우는 = 주변에 스페이스를 사용합니다.

  # Good
  def complex(real, imag=0.0):
    return magic(r=real, i=imag)
  

Bad

def complex(real, imag = 0.0): return magic(r = real, i = imag)

argument annotation with a default value

Good

def munge(sep: AnyStr = None): ... def munge(input: AnyStr, sep: AnyStr = None, limit=1000): ...

Bad

def munge(input: AnyStr=None): ... def munge(input: AnyStr, limit = 1000): ...

### Comment

- 코드와 반대되는 주석은 없는 것보다 못합니다.
- 첫번째 글자는 특수한 경우를 제외하고 대문자로 적도록 합니다.
- 마지막 문장을 제외하곤 문장의 끝에 두개의 스페이스를 사용하도록 합니다.
- 가능한 주석을 영어로 작성하도록 합니다.
- 인라인주석은 가능한 자제합니다.
- 인라인주석은 소스로부터 최소 2개의 스페이스로 구분되어야 합니다. 

### Documentation Strings

- 모든 public 모듈, 함수, 클래, 메소드에 대해 닥스트링을 작성합니다.
- Non-public 메소드에 대해서는 닥스트링이 필요하진 않지만, 메소드가 어떤 역할인지 설명하는 주석을 기재해야하고 이 주석은 def 라인 다음에 위치해야 한다.
- 여러줄로 구성된 닥스트링을 끝내는 마지막 따옴표는 혼자 있어야합니다. 만약 한줄로 작성된다면 같은 줄에 위치해도 됩니다.
``` python
# Google version
class Calculator:
    """
    Contains various functions to perform common 
    mathematical operations between two numbers

    Attributes:
        prevSum (int): Stores value of previous operation
    """

    def __init__(self):
        """
        Initializes class attributes

        Args:
            No arguments
        """

        self.prevSum = 0

    def add(num1, num2):
        """
        Calculate sum of two numbers

        Args:
            num1 (int): First Value
            num2 (int): Second Value

        Returns:
            Sum of num1 and num2
        """

        return num1 + num2

to be continue...

📌 현재 telegraf + influxDB + grafana의 조합으로 Airflow 메트릭을 모니터링하고 있습니다. 위 환경에서 이상 메트릭 감지 시, slack으로 alert을 보내기 위한 rule 정의 방법을 공유합니다.

Contact points

• alert을 보내는 목적지를 생성합니다.
• slack의 경우 token 혹은 Webhook URL로 채널을 지정할 수 있습니다.
• 커스텀 template을 사용할 경우, Text Body에 템플릿 name을 {{ template "airlfow_template" .}}과 같은 형태로 명시합니다.

Notification policies

• 이후에 생성할 rule이 어떤 point로 연결되는 지를 설정합니다.
• 여기서 작성한 label과 value를 rule 생성 시, 추가합니다.
• 연결하고자하는 point를 Contact point에서 지정합니다.

New alert rule

Rule type

• Rule name, Rule type을 지정해줍니다.
• folder의 경우 지정하지 않을 시 rule 생성이 불가한데 General folder가 리스트에 보이지 않아서 새로운 폴더를 생성해서 지정했습니다.

Create a query to be alerted on

• A section

  • 메트릭을 influxDB에 저장하고 있기 때문에 source를 influxDB로 지정해줍니다.
  • 각 노드(main, worker) 별로 memory use percent를 가지고 오는 쿼리를 작성해줍니다.

• B section

  • 각 노드별로 A에서 select한 메트릭 중 max 값을 추출합니다.

• C section

  • Math를 통해 표현식으로 B 결과를 filtering 합니다.
  • 위 예시는 75% 이상 사용하는 노드가 있는 경우를 식별합니다.

• Run queries

  • 실행 시, 작성한 조건에 맞는 결과를 조회할 수 있습니다.

Define alert conditions

• Condition에 alert을 위한 대상을 지정합니다.
• Evaluate은 alert의 기준 시간을 설정합니다. 1분 간격으로 측정하며 위에서 설정한 기준이 2분 동안 유지될 시, alert을 보냅니다.

Add details for your alert

• Summary and annotations에서 슬랙에 보낼 메세지를 지정할 수 있습니다.
• Custom Labels에 위 Notification policies에서 생성한 policy를 추가합니다.

Message templates

• 앞선 단계로도 충분하지만 slack에 메세지를 커스텀하고 싶을 경우, 사용할 수 있습니다.
• 위 예시는 실제 적용한 템플릿은 아니고 간단한 예제입니다.
• grafana에서 🗂default_template을 제공하고 있으니 템플릿 작성 시, 참고하시면 좋을 것 같습니다.

Filebeat란?

• Elastic Stack의 구성 요소 중 하나로, 로그 파일을 수집하고 전송하는 역할을 담당하는 경량 데이터 수집기입니다.
• Filebeat는 로그 파일, 이벤트 파일, 시스템 로그 등 다양한 유형의 로그 데이터를 모니터링하고, 이를 실시간으로 Elastic Stack의 다른 구성 요소로 전송합니다.

Filebeat 설치

• https://www.elastic.co/kr/downloads/past-releases/filebeat-7-10-2

설정파일

• elasticsearch log를 수집할 계획이므로 filebeat의 Elasticsearch module 을 이용해 log 파싱합니다.

filebeat.yml

filebeat.config.modules:
  path: ${MODULE_PATH}
  reload.enabled: false

setup.kibana:
  host: ${KIBANA_HOST}
  protocol: "http"
  dashboards.enabled: true

output.elasticsearch:
  hosts: [${ES_HOST}]

module yml

• 사용하려는 모듈 enable 후 해당 모듈 yml 작성합니다.

• enable 후 ${FILEBEAT_HOME}/modules.d 폴더 내 해당 모듈 yml이 생성됩니다.

• elasticsearch.yml

  # Module: elasticsearch
  # Docs: https://www.elastic.co/guide/en/beats/filebeat/7.10/filebeat-module-elasticsearch.html

• module: elasticsearch

  Server log

  server: enabled: true

  Set custom paths for the log files. If left empty,

  Filebeat will choose the paths depending on your OS.

  var.paths:

  - ${ES_LOG_PATH}/*_server.json

  audit: enabled: true

  Set custom paths for the log files. If left empty,

  Filebeat will choose the paths depending on your OS.

  var.paths:

  - ${ES_LOG_PATH}/*_audit.json

  slowlog: enabled: true

  Set custom paths for the log files. If left empty,

  Filebeat will choose the paths depending on your OS.

  var.paths:

  - ${ES_LOG_PATH}/*_index_search_slowlog.json
  - ${ES_LOG_PATH}/*_index_indexing_slowlog.json

  deprecation: enabled: true

  Set custom paths for the log files. If left empty,

  Filebeat will choose the paths depending on your OS.

  var.paths:

  - ${ES_LOG_PATH}/*_deprecation.json

Filebeat 서비스 활성화

# 서비스 활성화
systemctl enable filebeat.service

# 서비스 reload
systemctl daemon-reload

• /lib/systemd/system/filebeat.service를 작성합니다.

  Description=Filebeat sends log files to Logstash or directly to Elasticsearch.
  Documentation=https://www.elastic.co/products/beats/filebeat
  Wants=network-online.target
  After=network-online.target
  

[Service] ExecStart=${FILEBEAT_HOME}/filebeat -e -c ${FILEBEAT_HOME}/filebeat.yml Restart=always

[Install] WantedBy=multi-user.target

- service로 Filebeat를 실행합니다.
``` cmd
systemctl start filebeat

Airflow를 운영하다보면 마치 미역이 불어나듯 나날이 늘어나는 존재가 있습니다. 이놈을 간과했다간 어느새 log에 잠식당한 디스크를 볼 수 있을 겁니다.😱 이러한 참사를 막기 위해 적용한 log 최적화 방법을 정리하고자 합니다.

Dag log

• Worker에서 만들어내는 log로 각 DAG별로 생성됩니다.
• 해당 로그는 30일이 지난 log를 따로 삭제하는 DAG를 만들어 관리하고 있습니다.

  find $AIRFLOW_HOME/logs -type f -mtime +30 -delete

Scheduler log

• Scheduler에서 만들어내는 log로 airflow 컴포넌트 로그 중 가장 큰 용량을 차지하는 주범입니다.
• Scheduler와 Worker를 다른 노드에서 운영하고 있어 DAG를 이용한 삭제보다는 crontab으로 5일이 지난 log를 삭제하고 있습니다.

  find /var/lib/docker/overlay2/*/diff/opt/airflow/logs/scheduler -type d -mtime +5 -exec rm -rf {} +

Docker container log

• Docker container에서 만들어내는 log로 해당 로그는 Airflow의 문제가 아니라 Docker로 구성한 모든 시스템에 적용할 수 있습니다.
• 방법은 여러가지가 있지만 고려했던 3가지 방법을 소개합니다.

1. Logrotate

• Logrotate를 이용해 50MB를 초과하면 log 파일을 로테이트하고 파일이 5개 이상 쌓일 시, 삭제하도록 합니다.
• /etc/logrotate.d/docker에 아래 내용을 작성해줍니다.

  /var/lib/docker/containers/*/*.log {
   rotate 5
   size 50M
   compress
   missingok
   copytruncate
  }

2. docker-compose 설정

• docker-compose의 logging 설정을 통해 컨테이너별로 설정할 수 있습니다.
• 아래 예시는 50MB 이하의 log 파일을 최대 5개까지 유지합니다.

  airflow-worker:
    <<: *airflow-common
    container_name: airflow_worker
    hostname: ${_AIRFLOW_WORKER_HOSTNAME:-worker}
    command: celery worker
    volumes:
      ...
    healthcheck:
      test:
        - "CMD-SHELL"
        - 'celery --app airflow.executors.celery_executor.app inspect ping -d "celery@$${HOSTNAME}"'
      interval: 10s
      timeout: 10s
      retries: 5
    restart: always
    networks:
      - airflow
    logging:
      driver: 'json-file'
      options:
        max-size: '50m'
        max-file: '5'

3. docker 기본 설정

• docker 데몬에 logging driver 설정 시, 생성되는 모든 컨테이너에 동일하게 적용됩니다.
• /etc/docker/daemon.json에 아래 내용을 작성해줍니다.

  {
  "log-driver": "json-file",
  "log-opts": {
    "max-size": "50m",
    "max-file": "5"
  }
  }

오랜만에 복병을 만난 것 같지만... 차근차근 정리

쿠버네티스란?

• 컨테이너 오케스트레이션 도구
• k8s라는 표기는 “k”와 “s” 사이에 있는 8글자를 나타내는 약식 표기🧐
• 도커 스웜 모드처럼 여러 대의 도커 호스트를 하나의 클러스터로 만들어 준다는 점은 같지만 세부적인 기능을 도쿼스웜보다 많이 제공
• 장점
  • 서버 자원 클러스터링, 마이크로서비스 등 컨테이너 기반의 서비스 운영에 필요한 대부분의 오케스트레이션 기능 제공
  • 성능과 안정성 면에서 신뢰
  • 영속적 볼륨, 스케줄링, 장애 복구 등 컨테이너 기반의 클라우드를 운영할 때 필요한 대부분의 기능과 컴포넌트를 사용자가 직접 커스터마이징 할 수 있음
  • CNCF 및 다른 클라우드 운영도구들과 쉽게 연동
• 단점
  • 다른 오케스트레이션보다 다양한 지식 필요
  • 사용법이 복잡해 학습 비용이 큼

쿠버네티스 클러스터

• 컨트롤 플레인을 담당하는 마스터 노드와 애플리케이션 파트가 실행되는 워크노드로 구성

컨트롤 플레인

• 컨테이너 스케줄링, 서비스 관리, API 요청 처리 등의 작업을 수행 kube-apiserver : kubectl로부터 명령을 전달받아 실행 kube-scheduler : 새로운 POD 생성을 감지하고 실행시킬 워커를 선택 kube-controller-manager : 컨트롤러를 통합, 관리, 실행 cloud-controller-manager : 클라우드 서비스와 연동해 로드밸런서나 디스크 볼륨 같은 자원을 관리 etcd : 클러스터의 모든 데이터를 보관하는 일관성, 고가용성을 보장하는 키-값 저장소

노드 컴포넌트

• POD를 유지시키고 쿠버네티스 런타임 환경을 제공하는 역할을 수행 kubelet : 마스터의 kube-scheduler와 연동하여 워커 노드에 POD를 배치하고 실행하며 실행 중인 POD의 상태를 정기적으로 모니터링하며 kube-scheduler에 통지 kube-proxy : 각 노드에서 실행되는 네트워크 프록시

클러스터 설치

• 쿠버네티스 소프트웨어, CNI(가상 네트워크 드라이버)를 설치해야함
• CNI 소프트웨어는 플란넬, 칼리코, AWS VPC CNI 등이 있음
• 마스터 노드에는 etcd라는 데이터베이스, kubectl 설치
• 워커 노드에는 컨테이너 엔진 설치
• kubeadm, kops, GKE 등으로 클러스터 설치 가능
• *설치 시, 확인사항 *
  • 모든 서버의 시간이 ntp를 통해 동기화돼 있는지 확인
  • 모든 서버의 맥(MAC) 주소가 다른지 확인
  • 모든 서버가 메모리 2GB, 2CPU 이상의 자원을 가지고 있는지 확인
  • 모든 서버의 스왑을 비활성화. 메모리 스왑이 활성화돼 있으면 컨테이너의 성능이 일관되지 않을 수 있음 swapoff -a

Airflow는 batch 스케줄링을 위한 tool인만큼 사용하다보면 start_date, execution_date, schedule_interval등 날짜에 관련된 개념을 자주 마주하게 됩니다. 그러나 Airflow만의 독특한 날짜 개념도 있어 처음 사용할 때 혼동이 되는 경우가 많아 이를 간략하게 정리해보고자 합니다.

start_date

args = {
    "owner": "kyoon",
    "depends_on_past": False,
    "start_date": datetime(2022, 10, 27, tzinfo=local_tz),
    "end_date": datetime(2022, 12, 27, tzinfo=local_tz),
    "on_success_callback": alert.slack_success_alert,
    "on_failure_callback": alert.slack_fail_alert
}

• DAG 구동의 기준점이 되는 시간
• start_date는 한번 등록되면 DAG 파일에서 수정한다고 해도 업데이트 되지 않기 때문에 수정을 원할 시, DAG를 삭제한 후 새로 등록해야함
• end_date를 설정하면 DAG의 실행 중지 날짜를 지정할 수 있음
• DAG 클래스의 입력 파라미터 값들로 start_date(시작 시간)과 schedule_interval(실행 주기)를 입력하게 되는데 보통의 경우, 시작 시간부터 스케줄링이 시작된다고 생각하지만 airflow는 아래와 같이 작동되도록 설계

  DAG run = start_date + schedule_interval

• 예를 들어 start_date=datetime(2022.11.1), schedule_interval=@monthly의 스케줄을 가지는 DAG가 있고 실행시점은 11/27 5:48이라고 가정하면 이 DAG의 첫 실행일은 12/1 자정

schedule_interval

@dag(
    default_args=args,
    description="파이프라인",
    schedule_interval="30 2 * * *",
    ...
)

• DAG의 실행주기
• Airflow 매프로 프리셋 / Cron / 빈도 의 방법으로 스케줄을 정의할 수 있음

execution_date

• execution_date는 실행날짜가 아니라 주문번호(run id)에 가까움
• 한 가지 예를 들어 살펴보자. 매일 자정, 전날에 발생했던 log들을 es에 적재하는 작업이 있다고 가정한다면 자정이 지나 오늘의 날짜는 2022-12-28일인데 이관되어야 하는 log들의 날짜는 2022-12-27일이 된다. 이런 경우 사용자는 (오늘 날짜-1)을 하여 데이터를 가지고 오면 되는데 Airflow는 이를 execution_date라는 것으로 대신 사용할 수 게 해줌. 결국 실행되는 날짜는 2022-12-28이지만 execution_date는 2022-12-27이 됨.
• 나중에 해당 job을 다시 실행해도 execution_date는 그대로 유지.
  • 멱등성 유지
  • 다음날 job을 clear 후 재실행해도 execution_date는 처음 실행 때 할당된 값에서 변경되지 않음.
  • backfill 이나 task re-run 시, 유용
• Airflow에서는 job에서 사용되는 date값에 dynamic value인 now()보다 execution_date를 쓸 것을 권고.
• 주의할 점은, execution_date는 UTC(KST - 9h)를 기준으로 한다는 것.
  • KST : 2022-12-28 20:00:00 / UTC : 2022-12-28 11:00:00 / execution_date : 2022-12-27 11:00:00
  • KST : 2022-12-28 00:00:00 / UTC : 2022-12-27 15:00:00 / execution_date : 2022-12-26 15:00:00

execution_date 대신 쓸 수 있는 date 변수

• Airflow 공식문서를 보면 execution_date나 많이 쓰던 tomorrow_ds와 같은 변수는 사라질 예정이라고 함ㅠ.ㅠ
• 대안으로 쓸 수 있는 변수들은 아래 문서를 참고 https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html

참고

• https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html
• 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/

airflow를 사용하다보면 개발/운영 환경의 분리나 초기 적재/데일리 적재 등 옵션에 따라 다른 프로세스를 수행해야 할 때가 있다. 그때마다 dag나 task를 만들거나 직접 소스 내, config 파일을 바꾸는 것은 너무 비효율적이다. 이런 상황에서 사용할 수 있는 TIP이 있어 정리하고자 한다.

Trigger Dag w/ config

@dag(
    default_args=args,
    description="파이프라인",
    schedule_interval="30 3 * * *",
    catchup=False,
    tags=["data"],
    params={
        "mode" : "daily",
        "env" : "dev"
    }
)
def test_dag():
  ...

dag 정의 시, params라는 옵션을 입력한다. JSON형태로 key-value는 사용자가 정의 가능함.

• UI에서 Trigger DAG w/ config를 클릭하면 dag 내에서 지정한 params를 볼 수 있음
• params를 바꿔서 실행해야 한다면 화면에서 params를 바꾼 후 Trigger 버튼을 클릭하면 트리거 시점에 params를 변경할 수 있음
• 스케줄링 실행이나 기본 트리거 시에는 dag 내에 정의한 params가 반영됩니다.

airflow dags trigger --conf '{"mode":"dev", "env":"dev"}' test_dag

• CLI를 통해서 트리거 시에는 이와 같이 작성

params 사용법

위와 같이 지정한 params는 operator마다 사용방법이 다름

1. bash operator

   test_operator = BashOperator(
        task_id="test",
        bash_command="bash " + source_path + "/src/script/test.sh {{ params.env }}"
   )

• bash operator에서는 Jinja template 형태로 사용

2. python operator

   def choose_mode(**context):
    mode = context['params']['mode']
    if mode == 'init':
        target = 'init_generate_data'
    else:
        target = 'daily_generate_data'
    return target
   

task_branch = BranchPythonOperator( task_id='task_branch', python_callable=choose_mode )

- python operator에서는 context를 이용하여 전달받은 params를 읽을 수 있음

• 도커는 컨테이너 기반의 오픈소스 가상화 플랫폼
• 다양한 프로그램, 실행환경을 컨테이너로 추상화하고 동일한 인터페이스를 제공하여 프로그램의 배포 및 관리를 단순하게 해줌
• 리눅스 운영체제 위에서 동작

이미지

• 이미지는 컨테이너 실행에 필요한 파일과 설정값등을 포함하고 있는 것
• 도커 이미지는 Docker hub에 등록하거나 Docker Registry 저장소를 직접 만들어 관리할 수 있음
• 이미지는 url 방식으로 관리하며 태그를 붙일 수 있음

레이어

• 유니온 파일 시스템을 이용하여 여러개의 레이어를 하나의 파일시스템으로 사용할 수 있게 해줌
• 이미지는 여러개의 읽기 전용 레이어로 구성되고 파일이 추가되거나 수정되면 새로운 레이어가 생성
• 스토리지에 따라 레이어의 개수가 127개로 제한되어 있는 경우도 있음(명령어 최적화 필요)

Dockerfile

• 이미지를 만들기 위한 파일
• DSL(Domain-specific language) 언어를 이용
• 도커 레지스트리 등을 통해 배포할 때 이미지 대신 Dockerfile을 배포할 수 있음

1.기본 명령어

FROM

FROM <image>:<tag>
FROM ubuntu:16.04

• 베이스 이미지를 지정
• tag는 될 수 있으면 latest(기본값)보다 구체적인 버전(16.04등)을 지정하는 것이 좋음

MAINTAINER

MAINTAINER <name>
MAINTAINER subicura@subicura.com

• Dockerfile을 관리하는 사람의 이름 또는 이메일 정보

LABEL

LABEL <key>=<value>

• 이미지에 메타데이터를 추가할 때
• 키:값의 형태로 저장
• 추가된 메타데이터는 docker inspect 명령어로 확인

COPY

COPY <src>... <dest>
COPY . /usr/src/app

• 파일이나 디렉토리를 이미지로 복사
• 일반적으로 소스를 복사하는데 사용

ADD

ADD <src>... <dest>
ADD . /usr/src/app

• COPY와 유사
• src에 파일 대신 URL을 입력할 수 있고 src에 압축 파일을 입력하는 경우 자동으로 압축을 해제하면서 복사

RUN

RUN <command>
RUN ["executable", "param1", "param2"]
RUN bundle install

• 가장 많이 사용하는 구문(실행 명령어)
• 내부적으로 /bin/sh -c 뒤에 명령어를 실행하는 방식

CMD

CMD ["executable","param1","param2"]
CMD command param1 param2
CMD bundle exec ruby app.rb

• 컨테이너가 실행되었을 때 실행되는 명령어(빌드할 때는 실행X)
• 여러 개의 CMD가 존재할 경우 가장 마지막 CMD만 실행
• 한꺼번에 여러 개의 프로그램을 실행하고 싶은 경우에는 run.sh파일을 작성하여 데몬으로 실행하거나 supervisord나 forego와 같은 여러 개의 프로그램을 실행하는 프로그램을 사용

WORKDIR

WORKDIR /path/to/workdir

• RUN, CMD, ADD, COPY등이 이루어질 기본 디렉토리를 설정
• 각 명령어의 현재 디렉토리는 매번 초기화되기 때문에 RUN cd /path를 하더라도 다음 명령어에선 다시 위치가 초기화

EXPOSE

EXPOSE <port> [<port>...]
EXPOSE 4567

• 컨테이너가 실행되었을 때 요청을 기다리고 있는(Listen) 포트를 지정
• docker run 시, -P 옵션을 지정하면 이미지에 설정된 EXPOSE의 모든 포트를 호스트에 연결하도록 설정

VOLUME

VOLUME ["/data"]

• 컨테이너 외부에 파일시스템을 마운트 할 때 사용

ENV

ENV <key> <value>
ENV <key>=<value> ...
ENV DB_URL mysql

• 컨테이너에서 사용할 환경변수를 지정함
• run 명령어에서 -e 옵션을 사용할 경우 기존의 값은 overwrite

ARG

ARG <key>
ARG <key>=<value> ...
docker build --build-arg key=8080
CMD start.sh -h 127.0.0.1 -p ${key}

• build 명령어를 실행할 때 추가로 입력을 받아 Dockerfile 내에서 사용될 변수값을 설정

USER

USER <user_name>

• USER로 컨테이너 내부에서 사용될 사용자 계정의 이름을 설정하면 그 아래 명령어는 해당 사용자의 권한으로 실행
• 일반적으로 RUN으로 사용자의 그룹과 계정을 생성한 뒤 사용
• 루트 권한이 필요하지 않다면 USER를 사용하는 것을 권장

STOPSIGNAL

STOPSIGNAL SIGKIL

• 컨테이너가 정지될 때 사용될 시스템 콜의 종류 지정

HEALTHCHECK

HEALTHCHECK --interval=1m --timeout=3s --retries=3 {명령어}

• 애플리케이션의 상태를 체크하도록 하는 설정
• 상태체크에 실패하면 해당 컨테이너는 unhealthy 상태가 됨

SHELL

SHELL [경로]

• 사용하고자하는 셸을 따로 지정 시, 설정

ENTRYPOINT

ENTRYPOINT ["명령어"]

• CMD 명령어와 유사하지만 커멘드를 인자로 받아 사용할 수 있다는 점에서 차이가 있음
• ENTRYPOINT 가 설정되면 맨 마지막에 입력된 cmd를 인자로 받아 명령어를 출력
• CMD나 ENTRYPOINT 둘중 하나는 무조건 설정해야함
• 일반적으로 스크립트 파일을 인자로 입력해 컨테이너 실행 시, 해당 스크립트 파일을 실행하도록 함

2.Dockerfile 빌드

docker build -t {이미지명} {dockerfile 저장경로}

• -t 옵션을 사용하지 않으면 16진수 형태의 이름으로 이미지 저장

빌드과정

• 이미지 빌드를 시작하면 도커는 가장 먼저 빌드 컨텍스트를 읽음
  • 빌드 컨텍스트는 이미지를 생성하는데 필요한 파일, 소스코드, 메타데이터 등을 담고 있는 디렉터리
  • Dockerfile이 위치한 디렉터리가 빌드 컨텍스트가 됨
  • 해당 저장소에 위치한 모든 파일을 읽어들이므로 필요한 파일만 있는 것이 바람직
  • .dockerignore로 파일 제외 가능(Dockerfile과 같은 곳에 위치)
• 도커 빌드 과정
  • 임시 컨테이너 생성 -> 명령어 수행 -> 이미지 레이어로 저장 -> 임시 컨테이너 삭제 -> 새로 만든 이미지 기반 임시 컨테이너 생성 -> 명령어 수행 -> 이미지 레이어로 저장 -> 임시 컨테이너 삭제 -> …
  • 이미지 빌드가 완료되면 Dockerfile의 명령어 줄 수만큼의 레이어가 존재
• 이전에 빌드했던 Dockerfile 같은 내용이 있다면 캐시를 이용해 빌드
  • --no-cache 옵션을 통해 캐시 사용 선택 가능
  • --cache-from 옵션을 통해 캐시로 사용할 이미지 지정 가능

명령어 최적화

필요없는 로그 및 파일 무시

# before
RUN apt-get -y update

# after
RUN apt-get -y -qq update

• -qq 옵션으로 로그를 출력하지 않게 함

# before
RUN bundle install

# after
RUN bundle install --no-rdoc --no-ri

• --no-doc과 --no-ri 옵션으로 필요 없는 문서를 생성하지 않음

레이어 수 줄이기

# before
RUN apt-get -y -qq update
RUN apt-get -y -qq install ruby

# after
RUN apt-get -y -qq update && \
    apt-get -y -qq install ruby

• 비슷한 명령어끼리 묶어 레이어 수를 줄임
• 스토리지에 따라 레이어의 개수가 127개로 제한되어 있는 경우가 있음

• Pythonic 솔루션이 제시된 페이지를 참고해 구현 후 테스트 https://realpython.com/fast-flexible-pandas/
• 호텔 가격이 0~100까지 존재하고, 가격에 따라 high/medium/low 등급 측정
  • 0~29: low
  • 30~69: medium
  • 70~100: high

소스

# make test data
import numpy as np
import pandas as pd

rows = 1_000_000
hotel_price = pd.DataFrame([(f'hotel_{i}', int(p * 100)) for i, p in enumerate(np.random.rand(rows))], columns=['hotel', 'price'])
hotel_price

# 검증 source
level_all_algo = []
for algo in ['loop', 'iterrows', 'itertuples', 'withapply', 'isin', 'cut', 'digitize']:
    print(f'[{algo}]')
    %time globals()[f'apply_level_{algo}'](df=hotel_price)
    level_all_algo.append(hotel_price['level'].values)

import numpy as np
import pandas as pd
import sys

rows = 1_000_000
hotel_price = pd.DataFrame([(f'hotel_{i}', int(p * 100)) for i, p in enumerate(np.random.rand(rows))], columns=['hotel', 'price'])
hotel_price

def apply_level(price):
    if 0 <= price < 30:
        level = 'low'
    elif 30 <= price < 70:
        level = 'medium'
    elif 70 <= price <= 100:
        level = 'high'
    return level

def apply_level_loop(df: pd.DataFrame):
    level_list = []
    for i in range(len(df)):
        level = apply_level(df.iloc[i]['price'])
        level_list.append(level)
    df['level'] = level_list

def apply_level_iterrows(df: pd.DataFrame):
    level_list = []
    for i, r in df.iterrows():
        level = apply_level(r['price'])
        level_list.append(level)
    df['level'] = level_list

def apply_level_itertuples(df: pd.DataFrame):
    level_list = []
    for r in df.itertuples():
        level = apply_level(r.price)
        level_list.append(level)
    df['level'] = level_list

def apply_level_withapply(df: pd.DataFrame):
    df['level'] = df.apply(lambda r: apply_level(r['price']), axis=1)

def apply_level_isin(df: pd.DataFrame):
    low = df['price'].isin(range(0,30))
    medium = df['price'].isin(range(30,70))
    high = df['price'].isin(range(70,101))

    df.loc[low, 'level'] = 'low'
    df.loc[medium, 'level'] = 'medium'
    df.loc[high, 'level'] = 'high'

def apply_level_cut(df: pd.DataFrame):
    df['level'] = pd.cut(x=df['price'], bins=[0, 30, 70, 101], include_lowest=True, right=False, labels=['low','medium','high'])

def apply_level_digitize(df: pd.DataFrame):
    level_list = np.array(['low','medium','high'])
    bins = np.digitize(df['price'], bins=[30,70])
    df['level'] = level_list[bins]

level_all_algo = []
for algo in ['loop', 'iterrows', 'itertuples', 'withapply', 'isin', 'cut', 'digitize']:
    print(f'[{algo}]')
    %time globals()[f'apply_level_{algo}'](df=hotel_price)
    level_all_algo.append(hotel_price['level'].values)

np.array([(level_all_algo[0] == level).all() for level in level_all_algo]).all()
f'np: {np.__version__}, pd: {pd.__version__}, python: {sys.version}'        

결과

[loop]
CPU times: user 48.5 s, sys: 0 ns, total: 48.5 s
Wall time: 48.5 s
[iterrows]
CPU times: user 30 s, sys: 13.9 ms, total: 30 s
Wall time: 30 s
[itertuples]
CPU times: user 361 ms, sys: 7 µs, total: 361 ms
Wall time: 360 ms
[withapply]
CPU times: user 3.12 s, sys: 9.99 ms, total: 3.13 s
Wall time: 3.13 s
[isin]
CPU times: user 61.3 ms, sys: 14 µs, total: 61.3 ms
Wall time: 61.1 ms
[cut]
CPU times: user 21.5 ms, sys: 16 µs, total: 21.5 ms
Wall time: 24.7 ms
[digitize]
CPU times: user 41.4 ms, sys: 0 ns, total: 41.4 ms
Wall time: 41.4 ms
'np: 1.22.4, pd: 1.2.4, python: 3.8.0 (default, Dec  9 2021, 17:53:27) \n[GCC 8.4.0]'

결론

• 레퍼런스는 digitize이 cut보다 빠르다고 했지만 구현해보니 cut이 빨랐다...
• 가장 속도가 빨랐던 cut과 loop를 비교하니 2000배 가까이 빨랐다😱
• pandas에서 loop를 사용할 땐 웬만하면 digitize이나 cut을 사용할 것!

"details" : [
            {
              "value" : 0.0111732995,
              "description" : "score(freq=1.0), computed as boost * idf * tf from:",
              "details" : [
                {
                  "value" : 2.2,
                  "description" : "boost",
                  "details" : [ ]
                },
                {
                  "value" : 0.0111733,
                  "description" : "idf, computed as log(1 + (N - n + 0.5) / (n + 0.5)) from:",
                  "details" : [...]
                },
                {
                  "value" : 0.45454544,
                  "description" : "tf, computed as freq / (freq + k1 * (1 - b + b * dl / avgdl)) from:",
                  "details" : [...]
                }
              ]
            }
          ]

description을 확인해보면 아래의 수식을 찾을 수 있다.

boost * idf * tf

즉, score 계산은 boost(가중치 값)과 idf, tf를 곱한 값으로 이루어진다는 것을 알 수 있다. 그렇다면 idf와 tf는 뭘까? description을 다시 살펴보자.

Inverse Document Frequency(idf)

idf = log(1 + (N - n + 0.5) / (n + 0.5))

    N : 총 문서의 개수
    n : 문서에 단어가 등장하는 빈도수

수식을 보면 N값 대비 n값이 낮을수록 idf가 커진다.

즉, 문서에 해당 키워드가 등장하는 빈도가 작을수록, score가 높게. 키워드가 등장하는 빈도가 높을수록, score가 낮게 부여되는 방법이다.

Term Frequency(tf)

tfNorm = freq / (freq + k1 * (1 - b + b * dl / avgdl))

    freq : 문서에 나타나는 키워드 수
    k1 : 상수(elastic default 1.2)
    b : 상수(elastic default 0.75)
    dl : 검색된 필드의 길이
    avgdl : 평균 필드 길이

수식을 보면 freq(문서에 나타나는 키워드 수)이 높을수록, avgdl(평균 필드의 길이)대비 dl(검색된 문서의 필드 길이)가 짧을수록 tf의 크기가 커진다.

즉, 검색된 문서에 해당 키워드수가 자주 등장할 수록,평균 필드 길이보다 검색된 문서의 필드가 짧을수록 score가 높게 부여되는 방법이다.

결론

1. 전체 문서에서 검색 키워드가 등장하는 빈도수가 작을수록(즉, 유니크한 단어일수록)
2. 검색된 문서에서 검색 키워드가 많이 나타날수록
3. 평균 필드 길이보다 검색된 문서의 필드가 짧을수록 score가 올라간다!

설치 리스트

1. zookeeper

• 직접 애플리케이션 작업을 조율하지 않고 조율하는 것을 쉽게 개발할 수 있도록 도와주는 도구로 동기화나 마스터 선출 등의 작업을 쉽게 구현할 수 있게 해준다.

  version: '3'
  services:
    zookeeper:
      image: zookeeper:3.7
      hostname: zookeeper
      ports:
        - "2181:2181"
      environment:
        ZOO_MY_ID: 1
        ZOO_PORT: 2181
      volumes:
        - ./data/zookeeper/data:/data
        - ./data/zookeeper/datalog:/datalog

2. kafka cluster

• pub-sub 모델의 메세지큐

  version: '3'
  services:
    kafka1:
      image: confluentinc/cp-kafka:7.0.0
      hostname: kafka1
      ports:
        - "9091:9091"
      environment:
        KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka1:19091,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9091
        KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
        KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
        KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
        KAFKA_BROKER_ID: 1
        KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      volumes:
        - ./data/kafka1/data:/tmp/kafka-logs
      depends_on:
        - zookeeper

  environment

  # KAFKA_BROKER_ID
      클러스터 각 노드에 할당할 브로커의 아이디(중복불가) 
  # KAFKA_ZOOKEEPER_CONNECT 
      클러스터 구성에 사용할 주키퍼 호스트 정보
  # KAFKA_ADVERTISEMENT_LISTENERS
      카프카 브로커를 가리키는 주소 목록. 카프카 브로커는 초기 연결 시 이를 클라이언트에게 보낸다.
  # KAFKA_INTER_BROKER_LISTENER_NAME
      브로커 간 통신에 사용할 리스너명
  # KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
      보안 프로토콜 지정. PLAINTEXT는 리스너가 암호화되지 않은 것을 말함.
  # KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
      토픽 파티션의 복제 개수

3. kafdrop

• 카프카 클러스터를 관리하기 용이한 GUI

docker-compose.yml

• docker-compose.yml을 통해 설치
• kafka 클러스터보다 zookeeper가 먼저 실행되어야 함으로 파일 작성 시, zookeeper 컨테이너를 먼저 작성해준다.
• 로컬 환경에서 테스트를 목적으로 함으로 zookeeper는 싱글 노드, kafka는 3대의 브로커로 구성한다.

  version: '3'
  services:
  zookeeper:
    image: zookeeper:3.7
    hostname: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOO_MY_ID: 1
      ZOO_PORT: 2181
    volumes:
      - ./data/zookeeper/data:/data
      - ./data/zookeeper/datalog:/datalog
  kafka1:
    image: confluentinc/cp-kafka:7.0.0
    hostname: kafka1
    ports:
      - "9091:9091"
    environment:
      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka1:19091,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9091
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_BROKER_ID: 1
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    volumes:
      - ./data/kafka1/data:/tmp/kafka-logs
    depends_on:
      - zookeeper
  kafka2:
    image: confluentinc/cp-kafka:7.0.0
    hostname: kafka2
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka2:19092,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_BROKER_ID: 2
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    volumes:
      - ./data/kafka2/data:/tmp/kafka-logs
    depends_on:
      - zookeeper
  kafka3:
    image: confluentinc/cp-kafka:7.0.0
    hostname: kafka3
    ports:
      - "9093:9093"
    environment:
      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka3:19093,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_BROKER_ID: 3
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    volumes:
      - ./data/kafka3/data:/tmp/kafka-logs
    depends_on:
      - zookeeper
  kafdrop:
    image: obsidiandynamics/kafdrop
    restart: "no"
    ports: 
      - "9000:9000"
    environment:
      KAFKA_BROKER_CONNECT: "kafka1:19091"
    depends_on:
      - kafka1
      - kafka2
      - kafka3

  작성이 완료되었다면 아래와 같이 실행한다

  $ docker-compose -f docker-compose.yml up -d 

종료하는 방법은 아래를 참고한다.

$ docker-compose -f docker-compose.yal down

이후 kafka 컨테이너에 접속해 kafka 명령어를 사용할 수 있다.

$ docker exec -it ${컨테이너명} /bin/bash

혹은 kafdrop에 접속해(yml에 작성한 port로 접속) GUI로 kafka 클러스터를 관리할 수도 있다!

Catalyst

• 스파크 SQL의 질의 옵티마이저로 Logical Plan을 Physical Plan으로 바꾸는 일을 수행

  Logical Plan이란?

  • 수행해야하는 모든 transformation 단계에 대한 추상화
  • 데이터가 어떻게 변해야 하는지 정의하지만 실제 어디서 어떻게 동작 하는지는 정의하지 않음.

  Physical Plan이란?

  • Logical Plan이 어떻게 클러스터 위에서 실행될지 정의
  • 실행 전략을 만들고 Cost Model에 따라 최적화

  Catalyst의 실행 순서

  1. 분석
     • DataFrame 객체의 relation을 계산, 컬럼의 타입과 이름을 확인
  2. Logical Plan 최적화
     • 상수로 표현된 표현식을 compile time에 계산 (runtime 시 계산하지 않고)
     • Predicate Pushdown : join & filter -> filter & join
     • Project Pruning : 연산에 필요한 컬럼만 가져오기
  3. Physical Plan 만들기
     • 스파크에서 실행 가능한 Plan으로 변환
  4. 코드 제너레이션
     • 최적화된 Physical Plan을 Java Bytecode로 변환

Explain

• explain을 통해 plan을 조회할 수 있다

    spark.sql(query).explain(True)

1. Parsed Logical Plan
2. Analyzed Logical Plan
3. Optimized Logical Plan
4. Physical Plan

Tungsten

• Physical Plan이 선택되고 나면 분산환경에서 실행된 Bytecode가 만들어짐(코드 제너레이션)
• 스파크 엔진의 성능 향상이 목적
• 메모리 최적화 / 캐시 활용 연산 / 코드 생성

처음 Spark를 사용했던 프로젝트는 방대한 양의 금융데이터를 중복 연산해야하는 프로젝트였다. 데이터가 방대하다보니 초기 데이터를 로드하는 데만 해도 시간이 꽤 소요돼었다. 그때 Cache & Persist를 이용해 수행시간을 절약할 수 있었다.

Cache & Persist란?

• 데이터를 다루는 작업은 필연적으로 task가 반복되는 경우가 잦음.
• Spark의 지연되는 연산은 데이터를 메모리에 저장해둘 수 있어 유용. 이때 사용되는 것이 cache와 persist다.

1. cache

   • 디폴트 Storage Level을 사용
   • RDD : MEMORY_ONLY / DF : MEMORY_AND_DISK

2. Persist

   • Storage Level을 사용자가 원하는대로 지정 가능

    # cache()
    # cache()를 통해 categoryReviews 연산은 1번만 수행
    categoryReviews = filtered_lines.map(parse).cache()

    result1 = categoryReviews.take(10)
    result2 = categoryReviews.mapValues(lambda x: (x,1))

    # persist()
    from pyspark import StorageLevel
    categoryReviews = filtered_lines.map(parse).persist(StorageLevel.MEMORY_AND_DISK)

Storage Level

1. MEMORY_ONLY
2. MEMORY_AND_DISK
3. MEMORY_ONLY_SER
4. MEMORY_AND_DISK_SER
5. DISK_ONLY

** serialized 형태로 변환해서 저장해 저장 용량을 아낄 수 있지만 데이터 read시 재변환 연산을 수행해야함.

• Index란 DB의 테이블에 대한 검색 속도를 향상 시켜주는 자료구조.
• RDB는 데이터를 저장할 때는 내부적으로 아무런 순서없이 저장하는데 이때, 데이터 저장영역은 Heap 이라고 한다. Heap에서는 인덱스가 없는 테이블의 데이터를 찾을 때 전체 데이터 페이지의 처음 레코드부터 끝 페이지 마지막 레코드까지 모두 조회하게 된다.이러한 검색 방식을 풀 스캔(Full Scan) 또는 테이블 스캔(Table Scan)이라고 한다.

Index의 장점

1. 테이블을 검색하는 속도와 성능이 향상된다.(ORDER문이나 MIN/MAX도 마찬가지)
2. Index는 WHERE절에서 효과가 있다.

Index의 단점

1. 인덱스를 관리하기 위한 추가 작업 필요
2. 추가 저장공간 필요
3. 잘못 상용하는 경우 검색 성능 저하

• INSERT : 새로운 데이터에 대한 인덱스를 추가

• DELETE : 삭제하는 데이터의 인덱스를 사용하지 않는다는 작업 수행

• UPDATE : 기존의 인덱스를 사용하지 않음 처리, 갱신된 데이터에 대한 인덱스 추가

  이처럼 인덱스의 수정도 추가적으로 필요하기 때문에 데이터의 수정이 잦은 경우 성능이 낮아진다. 또 데이터의 인덱스를 제거하는 것이 아니라 '사용하지 않음'으로 처리하고 남겨두기 때문에 수정 작업이 많은 경우 실제 데이터에 비해 인덱스가 과도하게 커지는 문제점이 발생할 수 있다. 별도의 메모리 공간에 저장되기 때문에 추가 저장 공간도 많이 필요하게 된다. 

  또한 인덱스는 전체 데이터의 10 ~ 15% 이상의 데이터를 처리하거나, 데이터의 형식에 따라 오히려 성능이 낮아질 수 있다.예를 들어 나이나 성별과 같이 값의 range가 적은 컬럼인 경우, 인덱스를 읽고 나서 다시 많은 데이터를 조회해야 하기 때문에 비효율적이다. 

Index 설정 기준

앞서 설명한 단점으로 인해 마구잡이식의 인덱스 설정보다는 퍼포먼스를 극대화할 수 있는 설정 기준이 필요하다.

1. 카디널리티(Cardinality)가 높은 컬럼 카디널리티가 높다 = 한 컬럼이 가지고 있는 값의 중복도가 낮음(값들이 대부분 다른 값을 가짐) 카디널리티가 낮다 = 한 컬럼이 갖고 있는 값의 중복도가 높음(값들이 거의 같은 값을 가짐)

2. 선택도(Selectivity)가 낮은 컬럼 선택도가 5 ~ 10%가 적정 선택도가 높다 = 한 컬럼이 가지고 있는 값 하나로 여러 row가 찾아진다 선택도가 낮다 = 한 컬럼이 가지고 있는 값 하나로 적은 row가 찾아진다

    선택도 계산법 
    (= 컬럼의 특정 값의 row 수 / 테이블의 총 row 수 * 100)
    ex) 10개의 데이터에서 고유한 학번(grade) 컬럼, 2명씩 같은 이름(name) 컬럼, 5명씩 같은 나이(age) 컬럼인 경우
    ① 학번(grade) 컬럼 선택도: 1 / 10 = 10%
    ② 이름(name) 컬럼 선택도: 2 / 10 = 20%
    ③ 나이(age) 컬럼 선택도: 5 / 10 = 50%

3. 활용도가 높은 컬럼 WHERE, ORDER BY, JOIN에 자주 사용되는 컬럼에 사용

4. 수정빈도가 낮은 컬럼

Index를 타지 않는 쿼리

1. LIKE 연산자를 사용한 경우 - LIKE 연산자를 이용하여 검색할 경우 %를 뒤에 넣어 사용 - 가능하면 INSTR을 사용하는 것도 괜찮다.

   SELECT name FROM table WHERE name LIKE '%n%'
   -> SELECT name FROM table WHERE name LIKE 'n%'
   -> SELECT name FROM table WHERE INSTR(name, 'n') > 0

2. 수식이나 함수 등으로 인덱스 컬럼을 변형하였을 경우

3. 내부적으로 데이터 형 변환이 일어난 경우

  SELECT name FROM table WHERE name = '20200101'
  -> SELECT name FROM table WHERE name = TO_DATE('20200101', 'YYYYMMDD')

4. 조건절에 NULL 또는 NOT NULL을 사용하는 경우 - 인덱스를 구성한 컬럼값이 전부 NULL이라면 인덱스는 이런 값을 저장하지 않음

5. 부정형으로 조건을 사용한 경우

  SELECT name FROM table WHERE number != 10
  -> SELECT name FROM table WHERE number < 10 AND number > 10
  -> SELECT name FROM table WHERE NOT EXISTS (SELECT name FROM table WHERE number = 10)

6. OR 조건 사용 - UNION 절 활용

  SELECT t1.name FROM table t1, table2 t2 WHERE t1.name = t2.name OR t1.age = t2.age
  -> SELECT t1.name FROM table t1, table2 t2 WHERE t1.name = t2.name 
  UNION ALL
  SELECT t1.name FROM table t1, table2 t2 WHERE t1.age = t2.age 

결합 Index(Composite Index)

결합인덱스란 복수의 컬럼으로 구성된 인덱스

//결합 인덱스 생성 구문 예시
CREATE INDEX idx_composite_index
ON Test(id, name, date)

결합 Index의 컬럼 정하기

1. 조회조건에 항상 포함되는 경우

    - 결합인덱스이 첫번째 컬럼이 조건에 사용되지 않으면 그 결합인덱스는 선택되지 않음.

2. 첫번째 컬럼으로 최대한 많이 필터링할 수 있는 경우

    - 결합인덱스의 첫번째 컬럼 처리 범위를 산정하기 때문에 최대한 많은 필터링을 하도록 결정하는 것이 좋음. 

3. '=' 조건으로 사용되는 경우

       - 결합인덱스의 첫번째 컬럼이 '=' 조건이 아니라면 이후의 조건에서 '='을 사용하더라도 처리범위는 줄어들지 않음.

4. AND 조건으로 검색되는 경우

      - OR로 사용되는 경우는 결합인덱스를 사용하면 안됨.

메모리 계층 구조

CPU
L1 캐시
L2 캐시
L3 캐시
RAM
HDD/SSD

• 연산을 시작하면 하드디스크에서 CPU까지 데이터가 위로 이동
• 연산에 자주 쓰이는 데이터는 위로, 자주 쓰이지 않는 데이터는 아래로
• spark는 여러 노드에 분산 저장을 통해 in-memory 연산이 가능하도록 함.

SPARK 특징

1. in-memory 연산이 가능
2. Hadoop MapReduce보다 메모리 상에선 100배, 디스크 상에선 10배가 빠름
3. Transform + Action 으로 구성
4. 태스크를 정의할 때는 연산을 하지 않다가 결과가 필요할 때 연산 수행

SPARK 객체

1. SparkConf

• 사용자가 재정의해서 쓸 수 있는 설정 옵션들에 대한 키와 값을 갖고있는 객체

2. SparkContext

• Spark 클러스터와 연결시켜주는 객체
• Spark 모든 기능에 접근할 수 있는 시작점
• Spark는 분산환경에서 동작하기 때문에 Driver Program을 구동시키기 위해 SparkContext가 필요
• SparkContext는 프로그램당 하나만 만들 수 있고 사용 후에는 종료

1. config 폴더 아래 jvm.options 파일 수정

   -Xms8g -- 최소값
   -Xmx8g -- 최대값
   
   -XX:+UseG1GC
   -XX:MaxGCPauseMillis=200

2. 실행 시 환경 변수 이용 ** jvm.options 파일 내 옵션 주석 필요

    ./bin/elasticsearch ES_JAVA_OPTS="-Xms2g -Xmx2g"

설정 시, 주의사항

• 최소값과 최대값은 같게
• 힙의 최소값을 물리적 RAM의 50% 이상으로 설정하지 말 것.

설정 후, 재시작

• 이미 인덱스가 존재하고 있는 경우

  • cluster/settings의 allocation.enable=none으로 변경 해준 뒤, 다시 all로 변경
  • none으로 변경 후 node를 정지 -> 해당 node에 있던 primary shard가 다른 node에 있던 replica와 교환 -> 교환된 replica는 unassigned 상태로 변경 -> node 재기동 -> allocation.enable=all로 변경 -> unassigned replica가 재기동된 node에 할당

    PUT _cluster/settings
    {
    "transient" : {
        "cluster.routing.allocation.enable" : "none"
    }
    }
    

  //재시작 후 PUT _cluster/settings {

  "transient" : {
      "cluster.routing.allocation.enable" : "all"
  }

  }