When using the _DockerDecoratedOperator and python operator together, the celery executor throws a TypeError.
#35285
-
Apache Airflow version2.7.2 What happenedWhen using the docker operator and python operator together, the celery executor throws a TypeError. An error occurs when two operators are linked by a dependency. What you think should happen insteadNo response How to reproducein apache/airflow:slim-2.7.2-3.11 # test_dag.py
from __future__ import annotations
from os import environ
from airflow.decorators import dag, task
from pendulum.datetime import DateTime
from pendulum.tz import local_timezone
DEFAULT_ARGS = {
"image": "python:3.11-slim-bullseye",
"api_version": "auto",
"network_mode": "container:airflow-worker",
"docker_url": "TCP://docker-socket-proxy:2375",
"auto_remove": "force",
"mount_tmp_dir": False,
"container_name": "pickle_error_test",
"user": environ["AIRFLOW_UID"],
}
@task.python()
def no_error() -> None:
import logging
logger = logging.getLogger("airflow.task")
logger.info("in celery")
@task.docker()
def pickle_error() -> None:
import logging
logger = logging.getLogger("airflow.task")
logger.info("in docker")
@dag(
start_date=DateTime.now(local_timezone()).replace(
hour=0, minute=0, second=0, microsecond=0
),
schedule=None,
default_args=DEFAULT_ARGS | {"do_xcom_push": False},
catchup=False,
)
def test_docker_task_error() -> None:
in_celery = no_error()
in_docker = pickle_error()
# Removing the following line, no error occurs.
_ = in_celery >> in_docker
test_docker_task_error()Operating SystemUbuntu 22.04.1 LTS Versions of Apache Airflow Providersapache-airflow-providers-celery==3.3.4 DeploymentDocker-Compose Deployment detailsNo response Anything elseA pickle error will be thrown in executor, but the operator itself will run and exit normally(state: SUCCESS). Are you willing to submit PR?
Code of Conduct
|
Beta Was this translation helpful? Give feedback.
Replies: 3 comments 7 replies
-
|
Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval. |
Beta Was this translation helpful? Give feedback.
-
updateHow to reproducedocker-compose.yaml---
version: '3.8'
x-airflow-common:
&airflow-common
image: apache/airflow:2.7.2-python3.11
environment:
&airflow-common-env
AIRFLOW__CORE__EXECUTOR: CeleryExecutor
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@postgres/airflow
AIRFLOW__CELERY__BROKER_URL: redis://:@redis:6379/0
AIRFLOW__CORE__FERNET_KEY: ''
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
AIRFLOW__CORE__LOAD_EXAMPLES: 'true'
AIRFLOW__API__AUTH_BACKENDS: 'airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session'
AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK: 'true'
volumes:
- ./dags:/opt/airflow/dags
- ./logs:/opt/airflow/logs
- ./config:/opt/airflow/config
- ./plugins:/opt/airflow/plugins
user: ${AIRFLOW_UID}:0
depends_on:
&airflow-common-depends-on
redis:
condition: service_healthy
postgres:
condition: service_healthy
services:
postgres:
image: postgres:13
environment:
POSTGRES_USER: airflow
POSTGRES_PASSWORD: airflow
POSTGRES_DB: airflow
volumes:
- postgres-db-volume:/var/lib/postgresql/data
healthcheck:
test: ["CMD", "pg_isready", "-U", "airflow"]
interval: 10s
retries: 5
start_period: 5s
restart: always
redis:
image: redis:latest
expose:
- 6379
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 30s
retries: 50
start_period: 30s
restart: always
airflow-webserver:
<<: *airflow-common
command: webserver
ports:
- "8080:8080"
healthcheck:
test: ["CMD", "curl", "--fail", "http://localhost:8080/health"]
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-scheduler:
<<: *airflow-common
command: scheduler
healthcheck:
test: ["CMD", "curl", "--fail", "http://localhost:8974/health"]
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-worker:
<<: *airflow-common
command: celery worker
healthcheck:
# yamllint disable rule:line-length
test:
- "CMD-SHELL"
- 'celery --app airflow.providers.celery.executors.celery_executor.app inspect ping -d "celery@$${HOSTNAME}" || celery --app airflow.executors.celery_executor.app inspect ping -d "celery@$${HOSTNAME}"'
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
environment:
<<: *airflow-common-env
DUMB_INIT_SETSID: "0"
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-triggerer:
<<: *airflow-common
command: triggerer
healthcheck:
test: ["CMD-SHELL", 'airflow jobs check --job-type TriggererJob --hostname "$${HOSTNAME}"']
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-init:
<<: *airflow-common
entrypoint: /bin/bash
# yamllint disable rule:line-length
command:
- -c
- |
function ver() {
printf "%04d%04d%04d%04d" $${1//./ }
}
airflow_version=$$(AIRFLOW__LOGGING__LOGGING_LEVEL=INFO && gosu airflow airflow version)
airflow_version_comparable=$$(ver $${airflow_version})
min_airflow_version=2.2.0
min_airflow_version_comparable=$$(ver $${min_airflow_version})
if (( airflow_version_comparable < min_airflow_version_comparable )); then
echo
echo -e "\033[1;31mERROR!!!: Too old Airflow version $${airflow_version}!\e[0m"
echo "The minimum Airflow version supported: $${min_airflow_version}. Only use this or higher!"
echo
exit 1
fi
one_meg=1048576
mem_available=$$(($$(getconf _PHYS_PAGES) * $$(getconf PAGE_SIZE) / one_meg))
cpus_available=$$(grep -cE 'cpu[0-9]+' /proc/stat)
disk_available=$$(df / | tail -1 | awk '{print $$4}')
warning_resources="false"
if (( mem_available < 4000 )) ; then
echo
echo -e "\033[1;33mWARNING!!!: Not enough memory available for Docker.\e[0m"
echo "At least 4GB of memory required. You have $$(numfmt --to iec $$((mem_available * one_meg)))"
echo
warning_resources="true"
fi
if (( cpus_available < 2 )); then
echo
echo -e "\033[1;33mWARNING!!!: Not enough CPUS available for Docker.\e[0m"
echo "At least 2 CPUs recommended. You have $${cpus_available}"
echo
warning_resources="true"
fi
if (( disk_available < one_meg * 10 )); then
echo
echo -e "\033[1;33mWARNING!!!: Not enough Disk space available for Docker.\e[0m"
echo "At least 10 GBs recommended. You have $$(numfmt --to iec $$((disk_available * 1024 )))"
echo
warning_resources="true"
fi
if [[ $${warning_resources} == "true" ]]; then
echo
echo -e "\033[1;33mWARNING!!!: You have not enough resources to run Airflow (see above)!\e[0m"
echo "Please follow the instructions to increase amount of resources available:"
echo " https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html#before-you-begin"
echo
fi
mkdir -p /sources/logs /sources/dags /sources/plugins
chown -R "${AIRFLOW_UID}:0" /sources/{logs,dags,plugins}
exec /entrypoint airflow version
# yamllint enable rule:line-length
environment:
<<: *airflow-common-env
_AIRFLOW_DB_MIGRATE: 'true'
_AIRFLOW_WWW_USER_CREATE: 'true'
_AIRFLOW_WWW_USER_USERNAME: airflow
_AIRFLOW_WWW_USER_PASSWORD: airflow
_PIP_ADDITIONAL_REQUIREMENTS: ''
user: "0:0"
volumes:
- .:/sources
docker-socket-proxy:
image: tecnativa/docker-socket-proxy:0.1.1
container_name: airflow-socket
environment:
CONTAINERS: 1
IMAGES: 1
AUTH: 1
POST: 1
privileged: true
volumes:
- /var/run/docker.sock:/var/run/docker.sock:ro
restart: always
volumes:
postgres-db-volume:
dags/test_dag.py# test_dag.py
from __future__ import annotations
from os import environ
from airflow.decorators import dag, task
from pendulum.datetime import DateTime
from pendulum.tz import local_timezone
DEFAULT_ARGS = {
"image": "python:3.11-slim-bullseye",
"api_version": "auto",
"docker_url": "TCP://docker-socket-proxy:2375",
"auto_remove": "force",
"mount_tmp_dir": False,
"container_name": "pickle_error_test",
"user": environ["AIRFLOW_UID"],
}
@task.python()
def no_error() -> None:
import logging
logger = logging.getLogger("airflow.task")
logger.info("in celery")
@task.docker()
def pickle_error() -> None:
import logging
logger = logging.getLogger("airflow.task")
logger.info("in docker")
@dag(
start_date=DateTime.now(local_timezone()).replace(
hour=0, minute=0, second=0, microsecond=0
),
schedule=None,
default_args=DEFAULT_ARGS | {"do_xcom_push": False},
catchup=False,
)
def test_docker_task_error() -> None:
in_celery = no_error()
in_docker = pickle_error()
# Removing the following line, no error occurs.
_ = in_celery >> in_docker
test_docker_task_error() |
Beta Was this translation helpful? Give feedback.
-
|
This is at most troubleshooting. When docker task is used, the code to execute must be serialized to be able to send it to Docker container. Default For the future - please only open issues when you are reasonably sure you found a bug in Airflow, start with discussion if unsure. |
Beta Was this translation helpful? Give feedback.
-
|
@potiuk One more question, even if error occurs, the actual task is executed normally. Also, if docker operator is not |
Beta Was this translation helpful? Give feedback.
-
|
The error you see is from trying to schedule the "docker" task that's why you do not see it in the UI. It fails because "mini-scheduler" tries to scheduler the "docker" task right after execution of the in_celery task. It is configurable via this parameter (default is on): |
Beta Was this translation helpful? Give feedback.
-
Thank you @potiuk, I set the config and fixed the issue. Can you tell me which part of the script in example code is not serializable? Also, if I create a DockerOperator in a way that does the same thing as # test_dag2.py
from __future__ import annotations
import inspect
import os
import pickle
import textwrap
from os import environ
from tempfile import gettempdir
from airflow.decorators import dag, task
from airflow.providers.docker.decorators.docker import (
_b64_encode_file,
_generate_decode_command,
)
from airflow.providers.docker.operators.docker import DockerOperator
from airflow.utils.python_virtualenv import write_python_script
from pendulum.datetime import DateTime
from pendulum.tz import local_timezone
DEFAULT_ARGS = {
"image": "python:3.11-slim-bullseye",
"api_version": "auto",
"docker_url": "TCP://docker-socket-proxy:2375",
"auto_remove": "force",
"mount_tmp_dir": False,
"container_name": "pickle_error_test",
"user": environ["AIRFLOW_UID"],
}
@task.python()
def no_error() -> None:
import logging
logger = logging.getLogger("airflow.task")
logger.info("in celery")
def generate_command(python_command):
return (
f"""bash -cx '{_generate_decode_command("__PYTHON_SCRIPT", "/tmp/script.py",
python_command)} &&"""
f'{_generate_decode_command("__PYTHON_INPUT", "/tmp/script.in", python_command)} &&'
f"{python_command} /tmp/script.py /tmp/script.in /tmp/script.out'"
)
def get_python_source(python_callable):
raw_source = inspect.getsource(python_callable)
return textwrap.dedent(raw_source)
def pickle_error() -> None:
import logging
logger = logging.getLogger("airflow.task")
logger.warning("in docker")
@dag(
start_date=DateTime.now(local_timezone()).replace(
hour=0, minute=0, second=0, microsecond=0
),
schedule=None,
default_args=DEFAULT_ARGS | {"do_xcom_push": False},
catchup=False,
)
def test_docker_task_error2() -> None:
in_celery = no_error()
"""
airflow.providers.docker.decorators.docker._DockerDecoratedOperator.execute
"""
tmp_dir = gettempdir()
input_filename = os.path.join(tmp_dir, "script.in")
script_filename = os.path.join(tmp_dir, "script.py")
with open(input_filename, "wb") as file:
pickle.dump({"args": [], "kwargs": {}}, file)
py_source = get_python_source(pickle_error)
write_python_script(
jinja_context={
"op_args": [],
"op_kwargs": {},
"pickling_library": "pickle",
"python_callable": pickle_error.__name__,
"python_callable_source": py_source,
"expect_airflow": True,
"string_args_global": False,
},
filename=script_filename,
)
envs = {"__PYTHON_SCRIPT": _b64_encode_file(script_filename), "__PYTHON_INPUT": ""}
command = generate_command("python3")
in_docker = DockerOperator(
task_id="in_docker",
image="python:3.11-slim-bullseye",
environment=envs,
command=command,
)
# no error occurs.
_ = in_celery >> in_docker
test_docker_task_error2()
|
Beta Was this translation helpful? Give feedback.
-
|
The error occurs when copying the https://github.com/apache/airflow/blob/main/airflow/providers/docker/decorators/docker.py#L92 |
Beta Was this translation helpful? Give feedback.
-
|
logging module is not serializable as the message says. Again - if you want to serialize it - use dill (docker decorator has dill parameter for that) because you cannot serialize logging module with pickle. If you wish to understand more in detail what happens with serialization I suggest you experiment a bit and read dill/pickle documentation and self-explore the code. Serialization is a complex subject and using decorators is a bit magical so you have to deep dive and I do not know myself all details, enough to explain it in short message on github discussion. And you will not get all the answers straight away - also I suggest self-studying of the code and experimenting if you want to learn more - with tens of thousands of users we cannot focus on explainingsuch complex toipics like serialization in detail to individual users - that's a terrible waste of time when there are tens of PRs of people waiting to be commented and reviewed (which i heartily recomend you to do - you seem to be interested in detail so all I can ask you is you deep dive, self experiment, explore and start contributing back. Remember this is software created by volunteers so we help people here in our free time and contributing back is best way to learn how things work. I suggest you to mostly rely on self studying - but I think you got enough clues now. There a number of stack overflow questions about dill vs. pickle and you can also take a look at this https://github.com/uqfoundation/dill/blob/master/dill/_objects.py which explains which objects dill can serialize - there are also some nice tutorials out there like https://www.datacamp.com/tutorial/pickle-python-tutorial or pickle documentation https://docs.python.org/3/library/pickle.html |
Beta Was this translation helpful? Give feedback.
-
|
It's true that the I've confirmed that fixing the attribute that was causing the problem in |
Beta Was this translation helpful? Give feedback.
-
|
Yes. This is precisely why we are asking thos who are capable to dive deep, and they are able to find some errors maintainers might be not aware. I was under the impression there is a serialization of that module, but thanks for getting deeper and finding the root cause. Glad that you went that route. We will need a unit test for that change - comment in the PR. |
Beta Was this translation helpful? Give feedback.
It's true that the
loggingmodule can't serialize, but since_DockerDecoratedOperatorstores the original code of the decorated function as a string in a temporary file and passes it to docker, it should run fine as long as there are no dependencies on other variables inside the DAG file.I've confirmed that fixing the attribute that was causing the problem in
_DockerDecoratedOperatorfixes the issue, and I've requested a PR for it. Please review.#35293