Apache DolphinScheduler에서 작업자 작업 실행 분석

안녕하세요 여러분, 저는 WhaleOps의 수석 데이터 엔지니어인 Cai Shunfeng이고, Apache DolphinScheduler 커뮤니티의 커미터이자 PMC 멤버입니다. 오늘은 Apache DolphinScheduler의 Worker 작업이 어떻게 작동하는지 설명하겠습니다.

이 설명은 세 부분으로 나뉩니다.

1. Apache DolphinScheduler 소개
2. Apache DolphinScheduler의 전반적인 디자인 개요
3. Worker 작업의 세부 실행 프로세스

프로젝트 소개

Apache DolphinScheduler 는 엔터프라이즈 수준 시나리오에 적합한 분산형, 쉽게 확장 가능한 시각적 워크플로 스케줄링 오픈소스 시스템입니다.

이 솔루션은 시각적 작업을 통해 워크플로 및 작업에 대한 전체 라이프사이클 데이터 처리 솔루션을 제공하고 다음과 같은 주요 기능을 제공합니다.

주요 특징

• 사용하기 쉽습니다

• 시각적 DAG 작업: 사용자는 페이지에서 구성 요소를 끌어서 놓아 DAG(방향성 비순환 그래프)로 배열할 수 있습니다.

• 플러그인 시스템: 작업 플러그인, 데이터 소스 플러그인, 알림 플러그인, 스토리지 플러그인, 레지스트리 센터 플러그인, Cron 작업 플러그인 등이 포함됩니다. 사용자는 비즈니스 요구 사항을 충족시키기 위해 필요에 따라 플러그인을 쉽게 확장할 수 있습니다.

  
  

• 풍부한 사용 시나리오

• 정적 구성: 워크플로 스케줄링, 온라인 및 오프라인 작업, 버전 관리, 백필 기능이 포함됩니다.

• 런타임 작업: 일시 중지, 중지, 다시 시작 및 매개변수 대체와 같은 기능을 제공합니다.

• 종속성 유형: 다양한 종속성 옵션과 전략을 지원하여 더 많은 시나리오에 적응할 수 있습니다.

• 매개변수 전달: 워크플로 수준의 시작 매개변수, 글로벌 매개변수, 작업 수준의 로컬 매개변수 및 동적 매개변수 전달을 지원합니다.

  
  

• 높은 신뢰성

• 분산형 설계: 모든 서비스는 상태가 없으며 수평적으로 확장하여 시스템 처리량을 늘릴 수 있습니다.

• 과부하 보호 및 인스턴스 장애 허용:

• 과부하 보호: 작업 중에 마스터와 워커는 자체 CPU 및 메모리 사용량과 작업 볼륨을 모니터링합니다. 과부하가 발생하면 현재 워크플로/작업 처리를 일시 중지하고 복구 후 다시 시작합니다.

• 인스턴스 장애 허용: 마스터/워커 노드에 장애가 발생하면 레지스트리 센터는 서비스 노드가 오프라인임을 감지하고 워크플로 또는 작업 인스턴스에 대한 장애 허용을 수행하여 시스템의 자체 복구 기능을 최대한 보장합니다.

전반적인 디자인

프로젝트 아키텍처

다음으로, 전반적인 디자인 배경을 소개하겠습니다. 아래는 공식 웹사이트에 제공된 디자인 아키텍처 다이어그램입니다.

아키텍처 다이어그램에서 Apache DolphinScheduler가 여러 주요 구성 요소로 구성되어 있음을 알 수 있습니다.

• API 구성 요소: API 서비스는 주로 메타데이터를 관리하고, API 서비스를 통해 UI와 상호 작용하거나, API 인터페이스를 호출하여 워크플로 작업과 워크플로에 필요한 다양한 리소스를 생성합니다.

  
  

• 마스터 구성 요소: 마스터는 워크플로 인스턴스의 컨트롤러로, 명령을 사용하고, 이를 워크플로 인스턴스로 변환하고, DAG 분할을 수행하고, 작업을 순서대로 제출하고, 작업을 작업자에게 분배하는 역할을 합니다.

  
  

• Worker Component: Worker는 특정 작업의 실행자입니다. 작업을 수신한 후, 다양한 작업 유형에 따라 작업을 처리하고, 마스터와 상호 작용하고, 작업 상태를 보고합니다. 특히, Worker 서비스는 데이터베이스와 상호 작용하지 않습니다. API, 마스터, 알림 서비스만 데이터베이스와 상호 작용합니다.

  
  

• 알림 서비스: 알림 서비스는 다양한 알림 플러그인을 통해 알림을 보냅니다. 이러한 서비스는 레지스트리 센터에 등록하고 마스터와 워커는 정기적으로 하트비트와 현재 상태를 보고하여 정상적으로 작업을 받을 수 있도록 합니다.

마스터와 워커 상호작용 프로세스

마스터와 워커 간의 상호작용 과정은 다음과 같습니다.

• 작업 제출: 마스터는 DAG 분할을 완료한 후 데이터베이스에 작업을 제출하고 다양한 분산 전략에 따라 작업을 분산할 적절한 작업자 그룹을 선택합니다.

  
  

• 작업 수신: 작업자가 작업을 수신한 후, 작업의 상태에 따라 작업을 수락할지 여부를 결정합니다. 수락이 성공했는지 여부에 대한 피드백이 제공됩니다.

  
  

• 작업 실행: 작업자는 작업을 처리하고 상태를 실행 중으로 업데이트하고 마스터에 피드백합니다. 마스터는 데이터베이스에서 작업 상태 및 시작 시간 정보를 업데이트합니다.

  
  

• 작업 완료: 작업이 완료된 후, 워커는 마스터에 완료 이벤트 알림을 보내고 마스터는 ACK 확인을 반환합니다. ACK를 받지 못하면 워커는 작업 이벤트가 손실되지 않도록 계속 재시도합니다.

근로자 작업 접수

작업자가 작업을 받으면 다음과 같은 작업이 수행됩니다.

• 호스트 정보를 입력합니다.
• 작업자 머신에서 로그 경로를 생성합니다.
• 스레드 풀에 제출되어 실행될 Worker Task Executor를 생성합니다.

작업자는 과부하가 걸렸는지 확인하고, 과부하가 걸렸다면 작업을 거부합니다. 작업 분배 실패 피드백을 받은 후 마스터는 분배 전략에 따라 작업 분배를 위해 다른 작업자를 계속 선택합니다.

근로자 실행 프로세스

작업자 작업의 구체적인 실행 프로세스에는 다음 단계가 포함됩니다.

1. 작업 초기화: 작업에 필요한 환경과 종속성을 초기화합니다.
2. 작업 실행: 특정 작업 로직을 실행합니다.
3. 작업 완료: 작업 실행이 완료된 후, 작업 실행 결과를 마스터 노드에 보고합니다.

다음으로, 구체적인 작업 실행 프로세스를 자세히 설명하겠습니다.

작업 실행이 시작되기 전에 컨텍스트가 먼저 초기화됩니다. 이 시점에서 작업의 시작 시간이 설정됩니다. 작업의 정확성을 보장하려면 마스터와 워커 간의 시간을 동기화하여 시간 드리프트를 방지해야 합니다.

그런 다음 작업 상태가 실행 중으로 설정되고, 작업이 실행되기 시작했음을 알리기 위해 마스터로 피드백됩니다.

대부분의 작업이 Linux 운영 체제에서 실행되므로 테넌트 및 파일 처리가 필요합니다.

• 테넌트 처리: 먼저 테넌트가 존재하는지 확인합니다. 존재하지 않으면 구성에 따라 테넌트를 자동으로 생성할지 여부를 결정합니다. 이를 위해서는 배포 사용자가 작업 실행 중에 지정된 테넌트로 전환하기 위한 sudo 권한이 있어야 합니다.
• 특정 사용자 : 일부 시나리오에서는 테넌트를 전환할 필요가 없고, 특정 사용자를 사용하여 작업을 실행하기만 하면 됩니다. 이는 시스템에서도 지원됩니다.

• 
  

테넌트를 처리한 후, 워커는 특정 실행 디렉토리를 생성합니다. 실행 디렉토리의 루트 디렉토리는 구성 가능하며 적절한 권한이 필요합니다. 기본적으로 디렉토리 권한은 755로 설정됩니다.

작업 실행 중에 AWS S3 또는 HDFS 클러스터에서 파일을 가져오는 것과 같이 다양한 리소스 파일이 필요할 수 있습니다. 시스템은 이러한 파일을 후속 작업 사용을 위해 작업자의 임시 디렉토리에 다운로드합니다.

Apache DolphinScheduler에서 매개변수 변수는 대체될 수 있습니다. 주요 범주는 다음과 같습니다.

• 내장된 매개변수: 주로 시간 및 날짜 관련 매개변수를 대체하는 것과 관련됩니다.
• 사용자 정의 매개변수: 워크플로 또는 작업에서 사용자가 설정한 매개변수 변수도 그에 따라 대체됩니다.

위의 단계를 거쳐 작업의 실행 환경과 필요한 리소스가 준비되고, 작업은 공식적으로 실행을 시작할 수 있습니다.

다양한 유형의 작업

Apache DolphinScheduler에서는 다양한 유형의 작업이 지원되며, 각각 다른 시나리오와 요구 사항에 적용 가능합니다. 아래에서 몇 가지 주요 작업 유형과 해당 구성 요소를 소개합니다.

이러한 구성 요소는 다양한 스크립팅 언어 및 프로토콜에 적합한 스크립트 파일을 실행하는 데 일반적으로 사용됩니다.

• 셸: 셸 스크립트를 실행합니다.
• Python: Python 스크립트를 실행합니다.
• SQL: SQL 문을 실행합니다.
• 저장 프로시저: 데이터베이스에 저장된 프로시저를 실행합니다.
• HTTP: HTTP 요청을 수행합니다.

상업용 버전(WhaleScheduler)도 JAR 패키지를 실행하여 Java 애플리케이션을 실행하는 것을 지원합니다.

논리 작업 구성 요소

이러한 구성 요소는 논리적 제어 및 워크플로 관리를 구현하는 데 사용됩니다.

• 스위치: 조건 제어 작업.
• 종속성: 종속성 작업.
• 하위 프로세스: 하위 작업.
• NextLoop(상업용 버전): 금융 시나리오에 적합한 루프 제어 작업입니다.
• 트리거 구성 요소: 파일이나 데이터가 존재하는지 모니터링합니다.

빅데이터 구성 요소

이러한 구성 요소는 주로 빅데이터 처리 및 분석에 사용됩니다.

• SeaTunnel: 빅데이터 통합 및 처리에 사용되는 WhaleTunnel의 상용 버전에 해당합니다.
• AWS EMR: Amazon EMR 통합.
• HiveCli: Hive 명령줄 작업.
• Spark: Spark 작업.
• 플링크: 플링크 작업.
• DataX: 데이터 동기화 작업.

컨테이너 구성 요소

이러한 구성 요소는 컨테이너 환경에서 작업을 실행하는 데 사용됩니다.

• K8S: 쿠버네티스 작업.

데이터 품질 구성 요소

데이터 품질을 보장하는 데 사용됨:

• DataQuality: 데이터 품질 검사 작업입니다.

대화형 구성 요소

이러한 구성 요소는 데이터 과학 및 머신 러닝 환경과 상호 작용하는 데 사용됩니다.

• Jupyter: Jupyter Notebook 작업.
• 제플린: 제플린 노트북 작업.

머신 러닝 구성 요소

이러한 구성 요소는 머신 러닝 작업의 관리 및 실행에 사용됩니다.

• Kubeflow: Kubeflow 작업.
• MlFlow: MlFlow 작업.
• Dvc: 데이터 버전 제어 작업.

전반적으로 Apache DolphinScheduler는 스크립트 실행, 빅데이터 처리, 머신 러닝에 이르기까지 3~40개의 구성 요소를 지원합니다. 자세한 내용은 공식 웹사이트를 방문하여 자세한 설명서를 확인하세요.

작업 유형 추상화

Apache DolphinScheduler에서는 작업 유형이 여러 처리 모드로 추상화되어 다양한 런타임 환경과 요구 사항에 맞춥니다.

아래에서는 작업 유형의 추상화와 실행 프로세스를 자세히 소개합니다.

워커는 서버에 배포된 JVM 서비스입니다. 일부 스크립트 구성 요소(예: Shell, Python) 및 로컬 실행 작업(예: Spark Local)의 경우 별도의 프로세스를 시작하여 실행합니다.

이 시점에서 작업자는 프로세스 ID(PID)를 통해 이러한 작업과 상호 작용합니다.

다른 데이터 소스에는 다른 적응이 필요할 수 있습니다. SQL 및 저장 프로시저 작업의 경우 MySQL, PostgreSQL, AWS Redshift 등과 같은 다른 데이터 소스에 대한 처리를 추상화했습니다. 이 추상화를 통해 다양한 데이터베이스 유형의 유연한 적응 및 확장이 가능합니다.

원격 작업은 AWS EMR, SeaTunnel 클러스터, Kubernetes 클러스터 등과 같은 원격 클러스터에서 실행되는 작업을 말합니다. Worker는 이러한 작업을 로컬에서 실행하지 않고, 대신 원격 클러스터에 제출하고 상태와 메시지를 모니터링합니다. 이 모드는 확장성이 필요한 클라우드 환경에 특히 적합합니다.

작업 실행

로그 수집

플러그인마다 처리 모드가 다르므로 로그 수집도 그에 따라 달라집니다.

• 로컬 프로세스: 프로세스 출력을 모니터링하여 로그가 기록됩니다.

• 원격 작업: 원격 클러스터(예: AWS EMR)에서 작업 상태와 출력을 주기적으로 확인하여 로그를 수집하고 이를 로컬 작업 로그에 기록합니다.

  
  

매개변수 변수 대체

시스템은 동적으로 대체해야 하는 매개변수 변수를 식별하기 위해 작업 로그를 스캔합니다. 예를 들어, DAG의 작업 A는 다운스트림 작업 B에 전달해야 하는 일부 출력 매개변수를 생성할 수 있습니다.

이 과정에서 시스템은 로그를 읽고 필요에 따라 매개변수 변수를 대체합니다.

작업 ID 검색

• 로컬 프로세스: 프로세스 ID(PID)가 검색됩니다.
• 원격 작업: 원격 작업의 ID(예: AWS EMR 작업 ID)가 검색됩니다.

이러한 작업 ID를 보유하면 추가 데이터 쿼리와 원격 작업 작업이 가능합니다. 예를 들어, 워크플로가 중지되면 해당 취소 API를 작업 ID를 사용하여 호출하여 실행 중인 작업을 종료할 수 있습니다.

장애 허용 처리

• 로컬 프로세스: 작업자 노드가 실패하면 로컬 프로세스는 이를 인식하지 못해 작업을 다시 제출해야 합니다.
• 원격 작업: 작업이 원격 클러스터(예: AWS)에서 실행 중인 경우 작업 ID를 사용하여 작업 상태를 확인하고 작업을 인수하려고 시도할 수 있습니다. 성공하면 작업을 다시 제출할 필요가 없으므로 시간이 절약됩니다.

작업 실행 완료

작업이 실행된 후에는 여러 가지 완료 작업이 필요합니다.

• 작업 완료 확인: 시스템은 알림을 보내야 하는지 확인합니다. 예를 들어, SQL 작업의 경우 쿼리 결과가 알림을 트리거하면 시스템은 RPC를 통해 알림 서비스와 상호 작용하여 알림 메시지를 보냅니다.

• 이벤트 피드백: 작업자는 작업 완료 이벤트(완료 이벤트)를 마스터로 다시 보냅니다. 마스터는 데이터베이스에서 작업 상태를 업데이트하고 DAG 상태 전환을 진행합니다.

• 컨텍스트 정리: Worker는 작업 시작 시 생성된 작업 컨텍스트를 메모리에서 제거합니다. 또한 작업 실행 중에 생성된 파일 경로도 정리합니다. 디버그 모드(개발 모드)인 경우 이러한 파일은 정리되지 않으므로 실패한 작업의 문제 해결이 가능합니다.

  
  

이러한 단계를 거쳐 작업 인스턴스의 전체 실행 프로세스가 완료됩니다.

지역 사회 기여

Apache DolphinScheduler에 관심이 있으시고 오픈소스 커뮤니티에 기여하고 싶으시다면 기여 지침을 참조해주시기 바랍니다.

커뮤니티는 다음을 포함하되 이에 국한되지 않는 적극적인 기여를 장려합니다.

• 사용 중 발생한 문제를 보고합니다.
• 문서와 코드 PR을 제출합니다.
• 단위 테스트(UT) 추가.
• 코드 주석 추가.
• 버그 수정이나 새로운 기능 추가.
• 기술 문서 작성이나 미팅 참여.

신규 기여자를 위한 가이드

새로운 기여자의 경우 커뮤니티의 GitHub 이슈에서 good first issue 로 레이블이 지정된 이슈를 검색할 수 있습니다. 이러한 이슈는 일반적으로 더 간단하고 첫 번째 기여를 하는 사용자에게 적합합니다.

요약하면, 우리는 Apache DolphinScheduler의 전반적인 디자인과 Worker 작업의 자세한 실행 과정에 대해 알아보았습니다.

이 콘텐츠가 Apache DolphinScheduler를 더 잘 이해하고 사용하는 데 도움이 되기를 바랍니다. 질문이 있으면 댓글 섹션에서 저에게 연락하세요.