Task 실행관련 질문입니다.

Airflow 마스터 클래스

해결된 질문

작성

501

안녕하세요!

'Bash Operator & 외부 쉘파일 수행하기' 강의를 듣고 실습하던 중 에러가 생겨서, 해결을 시도했는데 계속해서 실패해서 질문을 남겨봅니다!

airflow를 실행하고 태스크를 수행하면 fail이 뜨는데, 로그를 확인하면 파일을 찾을 수 없다고 합니다.

 [2023-10-12, 14:12:25 KST] {subprocess.py:93} INFO - /bin/bash: line 1: /opt/***_log/plugins/shell/select_fruit.sh: No such file or directory
[2023-10-12, 14:12:25 KST] {subprocess.py:97} INFO - Command exited with return code 127
[2023-10-12, 14:12:25 KST] {taskinstance.py:1935} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/operators/bash.py", line 210, in execute
    raise AirflowException(
airflow.exceptions.AirflowException: Bash command failed. The command returned a non-zero exit code 127.

저는 wsl에 디렉토리는 아래 경로처럼 설정했습니다.(select_fruit.sh파일에 실행 권한도 주었습니다.)

/airflow_log/plugins/shell/select_fruit.sh

이후 docker-compose.yaml에 volumes 항목을 아래와 같이 설정을 했습니다.

 ${AIRFLOW_PROJ_DIR:-.}/airflow_log/plugins:/opt/airflow/plugins

Vscdoe에서 경로는 아래와 같이 설정을 했습니다.

t1_orange = BashOperator(
        task_id="t1_orange",
        bash_command="/opt/airflow_log/plugins/shell/select_fruit.sh ORANGE",
        
    )
    
    t2_banana = BashOperator(
        task_id="t2_banana",
        bash_command="/opt/airflow_log/plugins/shell/select_fruit.sh BANANA",
    )

경로 문제인지, 아니면 다른 문제인지.. 도움 요청해봅니다.

저에게 너무 필요한 강의라서 잘 듣고 있습니다! 앞으로도 좋은 강의 부탁드리겠습니다!

python 데이터-엔지니어링 airflow

답변 1

김현진

지식공유자

안녕하세요 pruione1님

task 실패한 이유는 docker-compose.yaml 파일의 volumes 항목에서 연결한 디렉토리와 dag이 바라보는 디렉토리가 다르기 때문입니다.

docker-compose.yaml 파일에는 아래와 같이 작성돼있죠?

 ${AIRFLOW_PROJ_DIR:-.}/airflow_log/plugins:/opt/airflow/plugins

따라서 WSL에서 작성한

$HOME/airlfow_log/plugins/shell/select_fruit.sh 디렉토리는

컨테이너에서 보면

/opt/airflow/plugins/shell/select_fruit.sh 에 위치하게 됩니다.

따라서 워커 컨테이너가 쉘 파일을 바라보도록 하기 위해 아래와 같이 수정해야 합니다.

bash_command="/opt/airflow_log/plugins/shell/select_fruit.sh ORANGE"
↓
bash_command="/opt/airflow/plugins/shell/select_fruit.sh ORANGE"

그럼 dag 내용은 이렇게 수정해야겠죠?

t1_orange = BashOperator(
        task_id="t1_orange",
        bash_command="/opt/airflow/plugins/shell/select_fruit.sh ORANGE",
        
    )
    
    t2_banana = BashOperator(
        task_id="t2_banana",
        bash_command="/opt/airflow/plugins/shell/select_fruit.sh BANANA",
    )

한번 해보시고 혹시 안되시면 또 남겨주세요 ^^

purione1

질문자

자세한 답변 감사합니다! 제가 잘못 이해하고 있었네요!

인프런 커뮤니티 질문&답변

Task 실행관련 질문입니다.