I'm trying to pass a pandas dataframe through one task to another using the Taskflow API paradigm.
Unlike in version 2.0.2, Airflow writes the tasks return values to the log files. As a result of this behaviour, my entire dataframe (84mb) is being written to a log file at every task execution.
- I am running Airflow in a Docker container using the 2.0.2 and 2.1.0 Docker image versions.
- I know this is not the proper way to pass heavy pieces of data through tasks.
I wanted to know if there is any configuration to avoid this new behaviour.
Here are some examples:
Task log in Airflow 2.0.2:
*** Reading local file: /opt/airflow/logs/procesador_tabla_ot_estructurada_nav/extraer/2021-06-15T11:43:56.695836+00:00/1.log
[2021-06-15 11:44:00,218] {taskinstance.py:877} INFO - Dependencies all met for <TaskInstance: procesador_tabla_ot_estructurada_nav.extraer 2021-06-15T11:43:56.695836+00:00 [queued]>
[2021-06-15 11:44:00,235] {taskinstance.py:877} INFO - Dependencies all met for <TaskInstance: procesador_tabla_ot_estructurada_nav.extraer 2021-06-15T11:43:56.695836+00:00 [queued]>
[2021-06-15 11:44:00,236] {taskinstance.py:1068} INFO -
--------------------------------------------------------------------------------
[2021-06-15 11:44:00,236] {taskinstance.py:1069} INFO - Starting attempt 1 of 1
[2021-06-15 11:44:00,236] {taskinstance.py:1070} INFO -
--------------------------------------------------------------------------------
[2021-06-15 11:44:00,245] {taskinstance.py:1089} INFO - Executing <Task(_PythonDecoratedOperator): extraer> on 2021-06-15T11:43:56.695836+00:00
[2021-06-15 11:44:00,251] {standard_task_runner.py:52} INFO - Started process 238 to run task
[2021-06-15 11:44:00,260] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'procesador_tabla_ot_estructurada_nav', 'extraer', '2021-06-15T11:43:56.695836+00:00', '--job-id', '22', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/procesador_tabla_ot_estructurada_nav.py', '--cfg-path', '/tmp/tmp23k2w273', '--error-file', '/tmp/tmpv3qqwlau']
[2021-06-15 11:44:00,262] {standard_task_runner.py:77} INFO - Job 22: Subtask extraer
[2021-06-15 11:44:00,325] {logging_mixin.py:104} INFO - Running <TaskInstance: procesador_tabla_ot_estructurada_nav.extraer 2021-06-15T11:43:56.695836+00:00 [running]> on host 25958216ea79
[2021-06-15 11:44:00,410] {taskinstance.py:1281} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=XXXXX
AIRFLOW_CTX_DAG_ID=procesador_tabla_ot_estructurada_nav
AIRFLOW_CTX_TASK_ID=extraer
AIRFLOW_CTX_EXECUTION_DATE=2021-06-15T11:43:56.695836+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2021-06-15T11:43:56.695836+00:00
[2021-06-15 11:44:26,166] {taskinstance.py:1185} INFO - Marking task as SUCCESS. dag_id=procesador_tabla_ot_estructurada_nav, task_id=extraer, execution_date=20210615T114356, start_date=20210615T114400, end_date=20210615T114426
[2021-06-15 11:44:26,225] {taskinstance.py:1246} INFO - 1 downstream tasks scheduled from follow-on schedule check
[2021-06-15 11:44:26,287] {local_task_job.py:146} INFO - Task exited with return code 0
Task log file in Airflow 2.1.0:
[2021-06-15 11:29:32,592] {taskinstance.py:876} INFO - Dependencies all met for <TaskInstance: procesador_tabla_ot_estructurada_nav.extraer 2021-06-15T11:29:30.550613+00:00 [queued]>
[2021-06-15 11:29:32,618] {taskinstance.py:876} INFO - Dependencies all met for <TaskInstance: procesador_tabla_ot_estructurada_nav.extraer 2021-06-15T11:29:30.550613+00:00 [queued]>
[2021-06-15 11:29:32,618] {taskinstance.py:1067} INFO -
--------------------------------------------------------------------------------
[2021-06-15 11:29:32,618] {taskinstance.py:1068} INFO - Starting attempt 1 of 1
[2021-06-15 11:29:32,618] {taskinstance.py:1069} INFO -
--------------------------------------------------------------------------------
[2021-06-15 11:29:32,627] {taskinstance.py:1087} INFO - Executing <Task(_PythonDecoratedOperator): extraer> on 2021-06-15T11:29:30.550613+00:00
[2021-06-15 11:29:32,633] {standard_task_runner.py:52} INFO - Started process 1110 to run task
[2021-06-15 11:29:32,637] {standard_task_runner.py:76} INFO - Running: ['***', 'tasks', 'run', 'procesador_tabla_ot_estructurada_nav', 'extraer', '2021-06-15T11:29:30.550613+00:00', '--job-id', '18', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/procesador_tabla_ot_estructurada_nav.py', '--cfg-path', '/tmp/tmpq89rbbgh', '--error-file', '/tmp/tmp99fa29i9']
[2021-06-15 11:29:32,639] {standard_task_runner.py:77} INFO - Job 18: Subtask extraer
[2021-06-15 11:29:32,711] {logging_mixin.py:104} INFO - Running <TaskInstance: procesador_tabla_ot_estructurada_nav.extraer 2021-06-15T11:29:30.550613+00:00 [running]> on host 6796ffa7a616
[2021-06-15 11:29:32,815] {taskinstance.py:1280} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=***
AIRFLOW_CTX_DAG_ID=procesador_tabla_ot_estructurada_nav
AIRFLOW_CTX_TASK_ID=extraer
AIRFLOW_CTX_EXECUTION_DATE=2021-06-15T11:29:30.550613+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2021-06-15T11:29:30.550613+00:00
**[2021-06-15 11:29:53,001] {python.py:151} INFO - Done. Returned value was: LITERALLY 84 MB OF A PANDAS DATAFRAME**
[2021-06-15 11:30:14,401] {taskinstance.py:1184} INFO - Marking task as SUCCESS. dag_id=procesador_tabla_ot_estructurada_nav, task_id=extraer, execution_date=20210615T112930, start_date=20210615T112932, end_date=20210615T113014
[2021-06-15 11:30:14,447] {taskinstance.py:1245} INFO - 1 downstream tasks scheduled from follow-on schedule check
[2021-06-15 11:30:14,508] {local_task_job.py:151} INFO - Task exited with return code 0