You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/05/11 11:56:37 UTC
[GitHub] [airflow] bhavaniravi opened a new issue, #23642: Dynamic Task Crashes scheduler - Non Empty Return
bhavaniravi opened a new issue, #23642:
URL: https://github.com/apache/airflow/issues/23642
### Apache Airflow version
2.3.0 (latest released)
### What happened
I have a dag that looks like this.
When I uncomment `py_job`(Dynamically mapped PythonOperator) it works well with `pull_messages` (Taskflow API).
When I try to do the same with `DatabricksRunNowOperator` it crashes the scheduler with error
### Sample DAG
```
import json
import pendulum
from airflow.decorators import dag, task
from airflow.operators.python import PythonOperator
from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator
@dag(
schedule_interval=None,
start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
catchup=False,
tags=['example'],
)
def tutorial_taskflow_api_etl():
def random(*args, **kwargs):
print ("==== kwargs inside random ====", args, kwargs)
print ("I'm random")
return 49
@task
def pull_messages():
return [["hi"], ["hello"]]
op = DatabricksRunNowOperator.partial(
task_id = "new_job",
job_id=42,
notebook_params={"dry-run": "true"},
python_params=["douglas adams", "42"],
spark_submit_params=["--class", "org.apache.spark.examples.SparkPi"]
).expand(jar_params=pull_messages())
# py_job = PythonOperator.partial(
# task_id = 'py_job',
# python_callable=random
# ).expand(op_args= pull_messages())
tutorial_etl_dag = tutorial_taskflow_api_etl()
```
### Error
```
[2022-05-11 11:46:30 +0000] [40] [INFO] Worker exiting (pid: 40)
return f(*args, **kwargs)
File "/usr/local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py", line 75, in scheduler
_run_scheduler_job(args=args)
File "/usr/local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py", line 46, in _run_scheduler_job
job.run()
File "/usr/local/lib/python3.9/site-packages/airflow/jobs/base_job.py", line 244, in run
self._execute()
File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 736, in _execute
self._run_scheduler_loop()
File "/usr/local/lib/python3.9/site-packages/astronomer/airflow/version_check/plugin.py", line 29, in run_before
fn(*args, **kwargs)
File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 824, in _run_scheduler_loop
num_queued_tis = self._do_scheduling(session)
File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 906, in _do_scheduling
callback_to_run = self._schedule_dag_run(dag_run, session)
File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 1148, in _schedule_dag_run
schedulable_tis, callback_to_run = dag_run.update_state(session=session, execute_callbacks=False)
File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
return func(*args, **kwargs)
File "/usr/local/lib/python3.9/site-packages/airflow/models/dagrun.py", line 522, in update_state
info = self.task_instance_scheduling_decisions(session)
File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
return func(*args, **kwargs)
File "/usr/local/lib/python3.9/site-packages/airflow/models/dagrun.py", line 658, in task_instance_scheduling_decisions
schedulable_tis, changed_tis, expansion_happened = self._get_ready_tis(
File "/usr/local/lib/python3.9/site-packages/airflow/models/dagrun.py", line 714, in _get_ready_tis
expanded_tis, _ = schedulable.task.expand_mapped_task(self.run_id, session=session)
File "/usr/local/lib/python3.9/site-packages/airflow/models/mappedoperator.py", line 609, in expand_mapped_task
operator.mul, self._resolve_map_lengths(run_id, session=session).values()
File "/usr/local/lib/python3.9/site-packages/airflow/models/mappedoperator.py", line 595, in _resolve_map_lengths
raise RuntimeError(f"Failed to populate all mapping metadata; missing: {keys}")
RuntimeError: Failed to populate all mapping metadata; missing: 'jar_params'
[2022-05-11 11:46:30 +0000] [31] [INFO] Shutting down: Master
```
### What you think should happen instead
_No response_
### How to reproduce
_No response_
### Operating System
Debian GNU/Linux 10 (buster)
### Versions of Apache Airflow Providers
apache-airflow-providers-databricks
### Deployment
Astronomer
### Deployment details
_No response_
### Anything else
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] jedcunningham commented on issue #23642: Dynamic Task Crashes scheduler - Non Empty Return
Posted by GitBox <gi...@apache.org>.
jedcunningham commented on issue #23642:
URL: https://github.com/apache/airflow/issues/23642#issuecomment-1129499616
Turns out it's related to whether the mapped field is a templated field or not. This is a simpler reproduction:
```python
from datetime import datetime
from airflow.decorators import dag, task
from airflow.operators.bash import BashOperator
class SimpleBashOperator(BashOperator):
template_fields = ()
@dag(schedule_interval=None, start_date=datetime(2021, 1, 1), catchup=False)
def non_templated_expand():
@task
def get_extra_env():
return [{"FOO": "bar"}, {"FOO": "biz"}]
SimpleBashOperator.partial(task_id="echo", bash_command="echo $FOO").expand(
env=get_extra_env()
)
non_templated_expand = non_templated_expand()
```
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ust40611 commented on issue #23642: Dynamic Task Crashes scheduler - Non Empty Return
Posted by GitBox <gi...@apache.org>.
ust40611 commented on issue #23642:
URL: https://github.com/apache/airflow/issues/23642#issuecomment-1132024680
@alex-astronomer Thanks for your workaround, do you mind suggest how could we pull_message() from xcom and add that message as a param to trigger the databricks job using DatabricksRunNowOperator?
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bhavaniravi commented on issue #23642: Dynamic Task Crashes scheduler - Non Empty Return
Posted by GitBox <gi...@apache.org>.
bhavaniravi commented on issue #23642:
URL: https://github.com/apache/airflow/issues/23642#issuecomment-1123704098
I did pass normal lists instead of task, it worked fine
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] jedcunningham closed issue #23642: Dynamic Task Crashes scheduler - Non Empty Return
Posted by GitBox <gi...@apache.org>.
jedcunningham closed issue #23642: Dynamic Task Crashes scheduler - Non Empty Return
URL: https://github.com/apache/airflow/issues/23642
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tirkarthi commented on issue #23642: Dynamic Task Crashes scheduler - Non Empty Return
Posted by GitBox <gi...@apache.org>.
tirkarthi commented on issue #23642:
URL: https://github.com/apache/airflow/issues/23642#issuecomment-1123665096
Can you please try with literal value `[["hi"], ["hello"]]` to `jar_params` instead of `pull_messages` ? On initial look this looks similar to https://github.com/apache/airflow/issues/23533
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] alex-astronomer commented on issue #23642: Dynamic Task Crashes scheduler - Non Empty Return
Posted by GitBox <gi...@apache.org>.
alex-astronomer commented on issue #23642:
URL: https://github.com/apache/airflow/issues/23642#issuecomment-1128123141
I've found that by adding a manual dependency, these tasks execute fine. Here is an example of that with the DAG given in the OP:
```
import json
import pendulum
from airflow.decorators import dag, task
from airflow.operators.python import PythonOperator
from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator
@dag(
schedule_interval=None,
start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
catchup=False,
tags=['example'],
)
def tutorial_taskflow_api_etl():
def random(*args, **kwargs):
print ("==== kwargs inside random ====", args, kwargs)
print ("I'm random")
return 49
@task
def pull_messages():
return [["hi"], ["hello"]]
pull_messages_task = pull_messages()
op = DatabricksRunNowOperator.partial(
task_id = "new_job",
job_id=42,
notebook_params={"dry-run": "true"},
python_params=["douglas adams", "42"],
spark_submit_params=["--class", "org.apache.spark.examples.SparkPi"]
).expand(jar_params=pull_messages_task)
pull_messages_task >> op
# py_job = PythonOperator.partial(
# task_id = 'py_job',
# python_callable=random
# ).expand(op_args= pull_messages())
tutorial_etl_dag = tutorial_taskflow_api_etl()
```
This appears to just be a problem with the way that task dependencies are automatically implemented for the task flow operator when using different operators. It depends on the operator that is being mapped. The dependencies are automatically created when mapping op_args in the PythonOperator, but the Databricks op used here is not doing that, and also a custom operator that I wrote is also not automagically creating the dependencies. Seems to be something specific to the operator that isn't implemented properly, if other operators inheriting from base are working fine.
```
class SumTwoMapOperator(PythonOperator):
def __init__(self, two_ints, *args, **kwargs):
super().__init__(*args, **kwargs)
self.python_operator = lambda: two_ints[0] + two_ints[1]
```
Let me know if you have any other questions here!
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org