Airflow 아키텍처 및 Executor들

def __init__(self, parallelism: int = PARALLELISM): super().__init__(parallelism=parallelism) ## BaseExecutor 상속 self.manager: Optional[SyncManager] = None self.result_queue: Optional['Queue[TaskInstanceStateType]'] = None self.workers: List[QueuedLocalWorker] = [] self.workers_used: int = 0 self.workers_active: int = 0 self.impl: Optional[ Union['LocalExecutor.UnlimitedParallelism', 'LocalExecutor.LimitedParallelism'] ] = None

self.impl: Optional[ Union['LocalExecutor.UnlimitedParallelism', 'LocalExecutor.LimitedParallelism'] ] = None

def start(self) -> None: """Start the executor.""" old_proctitle = getproctitle() setproctitle("airflow executor -- LocalExecutor") self.manager = Manager() setproctitle(old_proctitle) self.result_queue = self.manager.Queue() self.workers = [] self.workers_used = 0 self.workers_active = 0 self.impl = ( LocalExecutor.UnlimitedParallelism(self) if self.parallelism == 0 # 분기점 else LocalExecutor.LimitedParallelism(self) ) self.impl.start()

class UnlimitedParallelism: """ Implement LocalExecutor with unlimited parallelism, starting one process per command executed. :param executor: the executor instance to implement. """ def __init__(self, executor: LocalExecutor): self.executor: LocalExecutor = executor def start(self) -> None: """Start the executor.""" self.executor.workers_used = 0 self.executor.workers_active = 0

def execute_async( self, key: TaskInstanceKey, command: CommandType, queue: Optional[str] = None, executor_config: Optional[Any] = None, ) -> None: """ Executes task asynchronously. :param key: the key to identify the task instance :param command: the command to execute :param queue: Name of the queue :param executor_config: configuration for the executor """ if not self.executor.result_queue: raise AirflowException(NOT_STARTED_MESSAGE) local_worker = LocalWorker(self.executor.result_queue, key=key, command=command) self.executor.workers_used += 1 self.executor.workers_active += 1 local_worker.start()

def sync(self) -> None: """Sync will get called periodically by the heartbeat method.""" if not self.executor.result_queue: raise AirflowException("Executor should be started first") while not self.executor.result_queue.empty(): results = self.executor.result_queue.get() self.executor.change_state(*results) self.executor.workers_active -= 1

def change_state(self, key: TaskInstanceKey, state: TaskInstanceState, info=None) -> None: """ Change state of the task. :param info: Executor information for the task instance :param key: Unique key for the task instance :param state: State to set for the task. """ self.log.debug("Changing state: %s", key) try: self.running.remove(key) except KeyError: self.log.debug("Could not find key: %s", key) self.event_buffer[key] = state, info

class LocalWorkerBase(Process, LoggingMixin): """ LocalWorkerBase implementation to run airflow commands. Executes the given command and puts the result into a result queue when done, terminating execution. :param result_queue: the queue to store result state """ def __init__(self, result_queue: 'Queue[TaskInstanceStateType]'): super().__init__(target=self.do_work) # 프로세스 실행 self.daemon: bool = True self.result_queue: 'Queue[TaskInstanceStateType]' = result_queue def execute_work(self, key: TaskInstanceKey, command: CommandType) -> None: """ Executes command received and stores result state in queue. :param key: the key to identify the task instance :param command: the command to execute """ if key is None: return self.log.info("%s running %s", self.__class__.__name__, command) if settings.EXECUTE_TASKS_NEW_PYTHON_INTERPRETER: state = self._execute_work_in_subprocess(command) else: state = self._execute_work_in_fork(command) self.result_queue.put((key, state)) def _execute_work_in_subprocess(self, command: CommandType) -> str: try: subprocess.check_call(command, close_fds=True) return State.SUCCESS except subprocess.CalledProcessError as e: self.log.error("Failed to execute task %s.", str(e)) return State.FAILED ... @abstractmethod def do_work(self): """Called in the subprocess and should then execute tasks""" raise NotImplementedError()

Spawning a whole new python process and then re-loading all of Airflow is expensive. All though this time fades to insignificance for long running tasks, this delay gives a "bad" experience for new users when they are just trying out Airflow for the first time. For the LocalExecutor this cuts the "queued time" down from 1.5s to 0.1s on average.