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 2020/11/28 01:23:17 UTC
[GitHub] [airflow] mohamednizar opened a new issue #12676: Object of type 'DAG' is not JSON serializable
mohamednizar opened a new issue #12676:
URL: https://github.com/apache/airflow/issues/12676
**Apache Airflow version**:1.10.12
**What happened**:
I'm trying to generate dags based on config prams pass from trigger , But With all the implementation the DAG throwing an exception as `Object of type 'DAG' is not JSON serializable`
Please find my Dags code here with
```
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.models import DAG
from datetime import datetime, timedelta
from airflow.operators.mysql_operator import MySqlOperator
from airflow.hooks.mysql_hook import MySqlHook
from airflow.utils.dates import days_ago
import logging
dag = DAG(
dag_id='dynamic_superset_queries_dag_generator',
default_args={"owner": "airflow", "provide_context": True},
start_date=days_ago(1),
schedule_interval="@once"
)
start_task = DummyOperator(
task_id='start_task',
dag=dag
)
end = DummyOperator(
task_id='end',
dag=dag)
def generate_dags_for_queries(**context):
"""
access the payload params passed to the DagRun conf attribute.
:param context: The execution context
:type context: dict
"""
sql = format(context["dag_run"].conf["sql"])
table_name = format(context["dag_run"].conf["table_name"])
def insert_or_update_table():
"""
access the payload params passed to the DagRun conf attribute.
:param context: The execution context
:type context: dict
"""
try:
logging.info('trying the task')
logging.info('connecting to source')
src = MySqlHook(mysql_conn_id='openemis')
logging.info('connecting to destination')
print("Remotely received value of {} for key=sql".sql)
print("Remotely received value of {} for key=table_name".table_name)
dest = MySqlHook(mysql_conn_id='analytics')
src_conn = src.get_conn()
cursor = src_conn.cursor()
dest_conn = dest.get_conn()
cursor.execute(sql)
dest.insert_rows(table=table_name, rows=cursor, replace=True)
except Exception as e3:
logging.error('Table update is failed, please refer the logs more details')
logging.exception(context)
logging.exception(e3)
try:
table_name = format(context["dag_run"].conf["table_name"]).lower()
dag_name = f"dynamic_superset_queries_dag_{table_name}"
new_dag = DAG(
dag_name,
default_args={"owner": "airflow"},
start_date=days_ago(1),
schedule_interval=timedelta(minutes=5),
catchup=False
)
task_name = f"running_queries_{table_name}"
with new_dag:
dag_task = PythonOperator(
task_id=task_name,
python_callable=insert_or_update_table)
# Try to place the DAG into globals(), which doesn't work
globals()[dag_name] = new_dag
return dag
except Exception as e3:
logging.error('Dag creation failed , please refer the logs more details')
logging.exception(context)
logging.exception(e3)
with dag:
t1 = PythonOperator(
task_id='generate_dags_for_queries',
python_callable=generate_dags_for_queries,
dag=dag
)
start_task >> t1
t1 >> end
```
**What you expected to happen**:
It Should create DAGs with name of the table suffix
**How to reproduce it**:
I use run this command to trigger `airflow trigger_dag dynamic_superset_queries_dag_generator --conf '{"sql": "SELECT id, name, alternative_name,code,address FROM institutions LIMIT 1000", "table_name": "TEST_SCHOOLS"}'`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on issue #12676: Object of type 'DAG' is not JSON serializable
Posted by GitBox <gi...@apache.org>.
turbaszek commented on issue #12676:
URL: https://github.com/apache/airflow/issues/12676#issuecomment-735235340
@mohamednizar can you add full stack trace?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mohamednizar commented on issue #12676: Object of type 'DAG' is not JSON serializable
Posted by GitBox <gi...@apache.org>.
mohamednizar commented on issue #12676:
URL: https://github.com/apache/airflow/issues/12676#issuecomment-735351721
@mik-laj Shouldn't it included when we considering a bunch of same pattern of dags to be scheduled in deferment parameter ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb closed issue #12676: Object of type 'DAG' is not JSON serializable
Posted by GitBox <gi...@apache.org>.
ashb closed issue #12676:
URL: https://github.com/apache/airflow/issues/12676
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on issue #12676: Object of type 'DAG' is not JSON serializable
Posted by GitBox <gi...@apache.org>.
mik-laj commented on issue #12676:
URL: https://github.com/apache/airflow/issues/12676#issuecomment-735352457
I do not understand the question. Can you say a little more?
Documentation on dynamic DAGs;
http://apache-airflow-docs.s3-website.eu-central-1.amazonaws.com/docs/apache-airflow/latest/faq.html#how-can-i-create-dags-dynamically
New DAGs cannot be created in the PythonOperator because the DAG must exist in the module when it was loaded. This is needed for the Scheduler to be aware of the existence of this DAG in a given file.
This is how loading the DAG looks like.
First, the file is loaded and a new module is created
https://github.com/apache/airflow/blob/7ef9aa7d545f11442b6ebb86590cd8ce5f98430b/airflow/models/dagbag.py#L290-L295
All global variables in the module are checked
https://github.com/apache/airflow/blob/7ef9aa7d545f11442b6ebb86590cd8ce5f98430b/airflow/models/dagbag.py#L353
In the next step, when the DAG is loaded, it can be executed by the Worker or evaluated by the Scheduler
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on issue #12676: Object of type 'DAG' is not JSON serializable
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on issue #12676:
URL: https://github.com/apache/airflow/issues/12676#issuecomment-735023882
Thanks for opening your first issue here! Be sure to follow the issue template!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on issue #12676: Object of type 'DAG' is not JSON serializable
Posted by GitBox <gi...@apache.org>.
mik-laj commented on issue #12676:
URL: https://github.com/apache/airflow/issues/12676#issuecomment-735350307
Dynamic DAGs are not supported by Airflow. All DAG objects must exist each time a module is loaded.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on issue #12676: Object of type 'DAG' is not JSON serializable
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #12676:
URL: https://github.com/apache/airflow/issues/12676#issuecomment-785828836
Closing due to lack of activity. We can reopen this if we get more details.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org