You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Darwin Yip (Jira)" <ji...@apache.org> on 2020/02/12 18:18:00 UTC
[jira] [Updated] (AIRFLOW-6785) DagBag tries to run hook inside
SubDagOperator
[ https://issues.apache.org/jira/browse/AIRFLOW-6785?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Darwin Yip updated AIRFLOW-6785:
--------------------------------
Description:
The following worked in 1.10.1, but not in 1.10.9. It seems that the DagBag tries to execute the hook inside the SubdagOperator, which tries to connect to the database.
Assuming:
{code:python}
AIRFLOW_CONN_POSTGRES_CONN=postgresql+psycopg2://airflow:airflow@postgres:5432/airflow
{code}
{code:python}
import unittest
from airflow.models import DagBag
class TestDags(unittest.TestCase):
"""
Generic tests that all DAGs in the repository should be able to pass.
"""
LOAD_SECOND_THRESHOLD = 2
def setUp(self):
self.dagbag = DagBag()
def test_dagbag_import(self):
"""
Verify that Airflow will be able to import all DAGs in the repository.
"""
self.assertFalse(
len(self.dagbag.import_errors),
'There should be no DAG failures. Got: {}'.format(
self.dagbag.import_errors
)
)
{code}
{code:python}
from datetime import datetime, timedelta
from airflow import DAG
from airflow.hooks.postgres_hook import PostgresHook
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.operators.subdag_operator import SubDagOperator
default_args = {
"owner": "airflow",
"depends_on_past": False,
"start_date": datetime(2020, 2, 2),
"email": ["airflow@airflow.com"],
"email_on_failure": False,
"email_on_retry": False,
"retries": 1,
"retry_delay": timedelta(minutes=5),
}
def get_data(**kwargs):
df = PostgresHook(
postgres_conn_id=kwargs['postgres_conn_id']
).get_pandas_df("select 1;")
return df
def subdag(parent_dag_name, child_dag_name, args):
dag_subdag = DAG(
dag_id='%s.%s' % (parent_dag_name, child_dag_name),
default_args=args,
schedule_interval=None,
)
run_query = PythonOperator(
task_id=f'get_data_sub',
python_callable=get_data,
op_kwargs={
'postgres_conn_id': 'postgres_conn'
},
provide_context=True,
dag=dag_subdag
)
return dag_subdag
dag = DAG("test-hook-sub", default_args=default_args, schedule_interval=None)
start = DummyOperator(
task_id='kick_off',
dag=dag
)
section_1 = SubDagOperator(
task_id='section-1',
subdag=subdag("test-hook-sub", 'section-1', default_args),
dag=dag,
)
start >> section_1
{code}
Error:
{code:python}
psycopg2.OperationalError: could not translate host name "postgres" to address: Name or service not known
{code}
However the non-subdag version passes the test:
{code:java}
from datetime import datetime, timedelta
from airflow import DAG
from airflow.hooks.postgres_hook import PostgresHook
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
default_args = {
"owner": "airflow",
"depends_on_past": False,
"start_date": datetime(2020, 2, 2),
"email": ["airflow@airflow.com"],
"email_on_failure": False,
"email_on_retry": False,
"retries": 1,
"retry_delay": timedelta(minutes=5),
}
def get_data(**kwargs):
"""
Returns DB data as a Pandas DataFrame
"""
df = PostgresHook(
postgres_conn_id=kwargs['postgres_conn_id']
).get_pandas_df("select 1;")
return df
dag = DAG("test-hook", default_args=default_args, schedule_interval=None)
start = DummyOperator(
task_id='kick_off',
dag=dag
)
run_query = PythonOperator(
task_id=f'get_data',
python_callable=get_data,
op_kwargs={
'postgres_conn_id': 'postgres_conn'
},
provide_context=True,
dag=dag
)
start >> run_query
{code}
was:
The following worked in 1.10.1, but not in 1.10.9. It seems that the DagBag tries to execute the hook inside the SubdagOperator, which tries to connect to the database.
Assuming:
{code:python}
AIRFLOW_CONN_POSTGRES_CONN=postgresql+psycopg2://airflow:airflow@postgres:5432/airflow
{code}
{code:python}
import unittest
from airflow.models import DagBag
class TestDags(unittest.TestCase):
"""
Generic tests that all DAGs in the repository should be able to pass.
"""
LOAD_SECOND_THRESHOLD = 2
def setUp(self):
self.dagbag = DagBag()
def test_dagbag_import(self):
"""
Verify that Airflow will be able to import all DAGs in the repository.
"""
self.assertFalse(
len(self.dagbag.import_errors),
'There should be no DAG failures. Got: {}'.format(
self.dagbag.import_errors
)
)
{code}
{code:python}
from datetime import datetime, timedelta
from airflow import DAG
from airflow.hooks.postgres_hook import PostgresHook
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.operators.subdag_operator import SubDagOperator
default_args = {
"owner": "airflow",
"depends_on_past": False,
"start_date": datetime(2020, 2, 2),
"email": ["airflow@airflow.com"],
"email_on_failure": False,
"email_on_retry": False,
"retries": 1,
"retry_delay": timedelta(minutes=5),
}
def get_data(**kwargs):
df = PostgresHook(
postgres_conn_id=kwargs['postgres_conn_id']
).get_pandas_df("select 1;")
return df
def subdag(parent_dag_name, child_dag_name, args):
dag_subdag = DAG(
dag_id='%s.%s' % (parent_dag_name, child_dag_name),
default_args=args,
schedule_interval=None,
)
run_query = PythonOperator(
task_id=f'get_data_sub',
python_callable=get_data,
op_kwargs={
'postgres_conn_id': 'postgres_conn'
},
provide_context=True,
dag=dag_subdag
)
return dag_subdag
dag = DAG("test-hook-sub", default_args=default_args, schedule_interval=None)
start = DummyOperator(
task_id='kick_off',
dag=dag
)
section_1 = SubDagOperator(
task_id='section-1',
subdag=subdag("test-hook-sub", 'section-1', default_args),
dag=dag,
)
start >> section_1
{code}
Error:
{code:python}
psycopg2.OperationalError: could not translate host name "postgres" to address: Name or service not known
{code}
> DagBag tries to run hook inside SubDagOperator
> ----------------------------------------------
>
> Key: AIRFLOW-6785
> URL: https://issues.apache.org/jira/browse/AIRFLOW-6785
> Project: Apache Airflow
> Issue Type: Bug
> Components: DAG, hooks, operators
> Affects Versions: 1.10.9
> Reporter: Darwin Yip
> Priority: Major
>
> The following worked in 1.10.1, but not in 1.10.9. It seems that the DagBag tries to execute the hook inside the SubdagOperator, which tries to connect to the database.
> Assuming:
> {code:python}
> AIRFLOW_CONN_POSTGRES_CONN=postgresql+psycopg2://airflow:airflow@postgres:5432/airflow
> {code}
>
> {code:python}
> import unittest
> from airflow.models import DagBag
> class TestDags(unittest.TestCase):
> """
> Generic tests that all DAGs in the repository should be able to pass.
> """
> LOAD_SECOND_THRESHOLD = 2
> def setUp(self):
> self.dagbag = DagBag()
> def test_dagbag_import(self):
> """
> Verify that Airflow will be able to import all DAGs in the repository.
> """
> self.assertFalse(
> len(self.dagbag.import_errors),
> 'There should be no DAG failures. Got: {}'.format(
> self.dagbag.import_errors
> )
> )
> {code}
>
>
> {code:python}
> from datetime import datetime, timedelta
> from airflow import DAG
> from airflow.hooks.postgres_hook import PostgresHook
> from airflow.operators.dummy_operator import DummyOperator
> from airflow.operators.python_operator import PythonOperator
> from airflow.operators.subdag_operator import SubDagOperator
> default_args = {
> "owner": "airflow",
> "depends_on_past": False,
> "start_date": datetime(2020, 2, 2),
> "email": ["airflow@airflow.com"],
> "email_on_failure": False,
> "email_on_retry": False,
> "retries": 1,
> "retry_delay": timedelta(minutes=5),
> }
> def get_data(**kwargs):
> df = PostgresHook(
> postgres_conn_id=kwargs['postgres_conn_id']
> ).get_pandas_df("select 1;")
> return df
> def subdag(parent_dag_name, child_dag_name, args):
> dag_subdag = DAG(
> dag_id='%s.%s' % (parent_dag_name, child_dag_name),
> default_args=args,
> schedule_interval=None,
> )
> run_query = PythonOperator(
> task_id=f'get_data_sub',
> python_callable=get_data,
> op_kwargs={
> 'postgres_conn_id': 'postgres_conn'
> },
> provide_context=True,
> dag=dag_subdag
> )
> return dag_subdag
> dag = DAG("test-hook-sub", default_args=default_args, schedule_interval=None)
> start = DummyOperator(
> task_id='kick_off',
> dag=dag
> )
> section_1 = SubDagOperator(
> task_id='section-1',
> subdag=subdag("test-hook-sub", 'section-1', default_args),
> dag=dag,
> )
> start >> section_1
> {code}
> Error:
> {code:python}
> psycopg2.OperationalError: could not translate host name "postgres" to address: Name or service not known
> {code}
>
> However the non-subdag version passes the test:
> {code:java}
> from datetime import datetime, timedelta
> from airflow import DAG
> from airflow.hooks.postgres_hook import PostgresHook
> from airflow.operators.dummy_operator import DummyOperator
> from airflow.operators.python_operator import PythonOperator
> default_args = {
> "owner": "airflow",
> "depends_on_past": False,
> "start_date": datetime(2020, 2, 2),
> "email": ["airflow@airflow.com"],
> "email_on_failure": False,
> "email_on_retry": False,
> "retries": 1,
> "retry_delay": timedelta(minutes=5),
> }
> def get_data(**kwargs):
> """
> Returns DB data as a Pandas DataFrame
> """
> df = PostgresHook(
> postgres_conn_id=kwargs['postgres_conn_id']
> ).get_pandas_df("select 1;")
> return df
> dag = DAG("test-hook", default_args=default_args, schedule_interval=None)
> start = DummyOperator(
> task_id='kick_off',
> dag=dag
> )
> run_query = PythonOperator(
> task_id=f'get_data',
> python_callable=get_data,
> op_kwargs={
> 'postgres_conn_id': 'postgres_conn'
> },
> provide_context=True,
> dag=dag
> )
> start >> run_query
> {code}
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)