rustikk commented on issue #20974:
URL: https://github.com/apache/airflow/issues/20974#issuecomment-1048153501
Hey I set use_dill=True and I'm getting this traceback:
INFO - Executing cmd: /tmp/venv12se6ywa/bin/python
/tmp/venv12se6ywa/script.py /tmp/venv12se6ywa/script.in
/tmp/venv12se6ywa/script.out /tmp/venv12se6ywa/string_args.txt
[2022-02-22, 19:37:47 UTC] {process_utils.py:169} INFO - Output:
[2022-02-22, 19:37:48 UTC] {process_utils.py:173} INFO - Traceback (most
recent call last):
[2022-02-22, 19:37:48 UTC] {process_utils.py:173} INFO - File
"/tmp/venv12se6ywa/script.py", line 15, in <module>
[2022-02-22, 19:37:48 UTC] {process_utils.py:173} INFO - arg_dict =
dill.load(file)
[2022-02-22, 19:37:48 UTC] {process_utils.py:173} INFO - File
"/usr/local/lib/python3.9/site-packages/dill/_dill.py", line 270, in load
[2022-02-22, 19:37:48 UTC] {process_utils.py:173} INFO - return
Unpickler(file, ignore=ignore, **kwds).load()
[2022-02-22, 19:37:48 UTC] {process_utils.py:173} INFO - File
"/usr/local/lib/python3.9/site-packages/dill/_dill.py", line 472, in load
[2022-02-22, 19:37:48 UTC] {process_utils.py:173} INFO - obj =
StockUnpickler.load(self)
[2022-02-22, 19:37:48 UTC] {process_utils.py:173} INFO - File
"/usr/local/lib/python3.9/site-packages/dill/_dill.py", line 462, in find_class
[2022-02-22, 19:37:48 UTC] {process_utils.py:173} INFO - return
StockUnpickler.find_class(self, module, name)
[2022-02-22, 19:37:48 UTC] {process_utils.py:173} INFO -
ModuleNotFoundError: No module named
'unusual_prefix_50f8c09f995678d8614dafc9dd846f4d45c5416e_venv_xcoms'
This is the dag I'm using to get that traceback:
from platform import python_branch
from airflow.models import DAG
from airflow.operators.python import PythonVirtualenvOperator, PythonOperator
from airflow.utils.trigger_rule import TriggerRule
from airflow.utils.dates import days_ago
docs = """
####Purpose
The purpose of this dag is to check that xcoms works correctly with the
virtualenv operator.
It achieves this by making assertions that the data passed into xcoms
through return values is the same as what was defined.
Also, the dag parameter render_template_as_native_obj is set to True so the
comparison being made in the assertion can directly compare values, as xcom
serializes python return values into python strings.
####Expected Behavior
Both tasks are expected to succeed.
"""
from airflow_dag_introspection import log_checker
def push_some_data():
value = [1, "two", 3]
return value
def pull_that_data(**context):
pulled_data = context['ti'].xcom_pull(task_ids='push_data',
key="return_value")
print(type(pulled_data))
print(pulled_data)
#this a bug so turning it into a negative dag
#assert pulled_data is not None
#assert pulled_data[0] == [1, "two", 3]
with DAG(
dag_id="test_venv_op_xcoms",
start_date=days_ago(1),
schedule_interval="@once",
# mabye putting this in dag as this and checking that the virtualenv
package is installed
# as well as ensuring xcoms works are kind of like
PythonVirtualenvOperator sanity check tests
is_paused_upon_creation=False,
doc_md=docs,
tags=["core", "extended_tags", "venv_op"],
render_template_as_native_obj=True,
) as dag:
pusher = PythonVirtualenvOperator(
task_id="push_data",
python_callable=push_some_data,
)
puller = PythonVirtualenvOperator(
task_id="pull_data",
python_callable=pull_that_data,
#op_args=['{{ ti.xcom_pull(task_ids=["push_data"]) }}'],
use_dill=True,
)
pusher >> puller
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
