Question about updating python module in airflow-worker container

[sakaia]

Question about updating python module in airflow-worker container

I deploy airflow by helm. and I am trying to execute airflow-worker with version 1.10.14.
The package requires marshmallow package updates by pip. but it does not reflect in python code.
Is there any way to solve this issue?

# Airflow marshmallow version update test

from airflow.utils.dates import days_ago
from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator

af_run_dag_default_args = {
    'owner': 'sakaia'
}

## Define DAG details
af_run_dag = DAG(
    dag_id='ss0',
    default_args=af_run_dag_default_args,
    schedule_interval=None,
    start_date=days_ago(2),
    tags=['training']
)

def pythonTest(**kwargs) -> str :
    print("pythonTest")
    import sys, subprocess
    result = subprocess.check_output([sys.executable, '-m', 'pip', 'install', '--user', 'marshmallow==3.10.0'])
    print(str(result).replace('\\n', '\n'))

    import marshmallow
    print("marshmallow version: %s" % marshmallow.__version__, file=sys.stderr)
    print(str("marshmallow version %s" % marshmallow.__version__))
    print(str("marshmallow file    %s" % marshmallow.__file__))
    with open('/home/airflow/.local/lib/python3.8/site-packages/marshmallow/__init__.py') as f:
       hoge = f.read()
    print(hoge)
    return "pythonTest"

# Define DAG steps/workflow
with af_run_dag as dag :
    python_test = PythonOperator(
         task_id='python_test',
         python_callable=pythonTest,
         op_kwargs={
         },
         dag=dag
    )

Output log is follows, it seems marshmallow version update works successfully, but it keeps version 2.21.0 on python code.

[2021-02-05 01:45:36,385] {base_task_runner.py:61} DEBUG - Planning to run as the  user
[2021-02-05 01:45:36,415] {taskinstance.py:685} DEBUG - <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [queued]> dependency 'Task Instance State' PASSED: True, Task state queued was valid.
[2021-02-05 01:45:36,415] {taskinstance.py:685} DEBUG - <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [queued]> dependency 'Not In Retry Period' PASSED: True, The task instance was not marked for retrying.
[2021-02-05 01:45:36,422] {taskinstance.py:685} DEBUG - <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [queued]> dependency 'Trigger Rule' PASSED: True, The task instance did not have any upstream tasks.
[2021-02-05 01:45:36,423] {taskinstance.py:685} DEBUG - <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [queued]> dependency 'Previous Dagrun State' PASSED: True, The task did not have depends_on_past set.
[2021-02-05 01:45:36,423] {taskinstance.py:670} INFO - Dependencies all met for <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [queued]>
[2021-02-05 01:45:36,442] {taskinstance.py:685} DEBUG - <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [queued]> dependency 'Pool Slots Available' PASSED: True, ('There are enough open slots in %s to execute the task', 'default_pool')
[2021-02-05 01:45:36,443] {taskinstance.py:685} DEBUG - <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [queued]> dependency 'Task Concurrency' PASSED: True, Task concurrency is not set.
[2021-02-05 01:45:36,443] {taskinstance.py:685} DEBUG - <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [queued]> dependency 'Not In Retry Period' PASSED: True, The task instance was not marked for retrying.
[2021-02-05 01:45:36,443] {taskinstance.py:685} DEBUG - <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [queued]> dependency 'Previous Dagrun State' PASSED: True, The task did not have depends_on_past set.
[2021-02-05 01:45:36,468] {taskinstance.py:685} DEBUG - <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [queued]> dependency 'Trigger Rule' PASSED: True, The task instance did not have any upstream tasks.
[2021-02-05 01:45:36,472] {taskinstance.py:670} INFO - Dependencies all met for <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [queued]>
[2021-02-05 01:45:36,472] {taskinstance.py:880} INFO -
--------------------------------------------------------------------------------
[2021-02-05 01:45:36,472] {taskinstance.py:881} INFO - Starting attempt 1 of 1
[2021-02-05 01:45:36,472] {taskinstance.py:882} INFO -
--------------------------------------------------------------------------------
[2021-02-05 01:45:36,506] {taskinstance.py:901} INFO - Executing <Task(PythonOperator): python_test> on 2020-12-07T00:00:00+00:00
[2021-02-05 01:45:36,512] {standard_task_runner.py:54} INFO - Started process 70 to run task
[2021-02-05 01:45:36,562] {standard_task_runner.py:77} INFO - Running: ['airflow', 'run', 'ss0', 'python_test', '2020-12-07T00:00:00+00:00', '--job_id', '87', '--pool', 'default_pool', '--raw', '-sd', 'DAGS_FOLDER/ss0.py', '--cfg_path', '/tmp/tmp0wif1hiv']
[2021-02-05 01:45:36,562] {standard_task_runner.py:78} INFO - Job 87: Subtask python_test
[2021-02-05 01:45:36,563] {cli_action_loggers.py:68} DEBUG - Calling callbacks: [<function default_action_log at 0x7f166678ab80>]
[2021-02-05 01:45:36,587] {settings.py:233} DEBUG - Setting up DB connection pool (PID 70)
[2021-02-05 01:45:36,588] {settings.py:267} DEBUG - settings.prepare_engine_args(): Using NullPool
[2021-02-05 01:45:36,652] {logging_mixin.py:112} INFO - Running <TaskInstance: ss0.python_test 2020-12-07T00:00:00+00:00 [running]> on host ss0pythontest-ec131a89bec94531956e82eb98c543c1
[2021-02-05 01:45:36,726] {__init__.py:101} DEBUG - Preparing lineage inlets and outlets
[2021-02-05 01:45:36,726] {__init__.py:137} DEBUG - inlets: [], outlets: []
[2021-02-05 01:45:36,727] {python_operator.py:103} DEBUG - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=sakaia
AIRFLOW_CTX_DAG_ID=ss0
AIRFLOW_CTX_TASK_ID=python_test
AIRFLOW_CTX_EXECUTION_DATE=2020-12-07T00:00:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=backfill_2020-12-07T00:00:00+00:00
[2021-02-05 01:45:36,727] {logging_mixin.py:112} INFO - pythonTest
[2021-02-05 01:45:40,690] {logging_mixin.py:112} INFO - b'Collecting marshmallow==3.10.0
  Downloading marshmallow-3.10.0-py2.py3-none-any.whl (46 kB)
Installing collected packages: marshmallow
  Attempting uninstall: marshmallow
    Found existing installation: marshmallow 2.21.0
    Uninstalling marshmallow-2.21.0:
      Successfully uninstalled marshmallow-2.21.0
Successfully installed marshmallow-3.10.0
'
[2021-02-05 01:45:40,692] {logging_mixin.py:112} WARNING - marshmallow version: 2.21.0
[2021-02-05 01:45:40,692] {logging_mixin.py:112} INFO - marshmallow version 2.21.0
[2021-02-05 01:45:40,692] {logging_mixin.py:112} INFO - marshmallow file    /home/airflow/.local/lib/python3.8/site-packages/marshmallow/__init__.py
[2021-02-05 01:45:40,693] {logging_mixin.py:112} INFO - from marshmallow.schema import Schema, SchemaOpts

from . import fields
from marshmallow.decorators import (
    pre_dump,
    post_dump,
    pre_load,
    post_load,
    validates,
    validates_schema,
)
from marshmallow.utils import EXCLUDE, INCLUDE, RAISE, pprint, missing
from marshmallow.exceptions import ValidationError
from distutils.version import LooseVersion

__version__ = "3.10.0"
__version_info__ = tuple(LooseVersion(__version__).version)
__all__ = [
    "EXCLUDE",
    "INCLUDE",
    "RAISE",
    "Schema",
    "SchemaOpts",
    "fields",
    "validates",
    "validates_schema",
    "pre_dump",
    "post_dump",
    "pre_load",
    "post_load",
    "pprint",
    "ValidationError",
    "missing",
]
[2021-02-05 01:45:40,693] {logging_mixin.py:112} INFO -
[2021-02-05 01:45:40,693] {python_operator.py:114} INFO - Done. Returned value was: pythonTest
[2021-02-05 01:45:40,736] {__init__.py:61} DEBUG - Backend: None, Lineage called with inlets: [], outlets: []
[2021-02-05 01:45:40,766] {taskinstance.py:1057} INFO - Marking task as SUCCESS.dag_id=ss0, task_id=python_test, execution_date=20201207T000000, start_date=20210205T014536, end_date=20210205T014540
[2021-02-05 01:45:41,396] {base_job.py:197} DEBUG - [heartbeat]
[2021-02-05 01:45:41,396] {local_task_job.py:102} INFO - Task exited with return code 0