Dynamic DAGs and Pandas to_sql behaviour

[ziggekatten]

First of all, I'm quite new to Airflow, but have been using it for some time, and have built a few DAGs handling various stuff, as well as building an custom xcom backend.

Now, here is an scenario that baffels me. I have five sources that should be treated in te same way, thwe only thing differing is the name, and the end destination, in this case Postgres tables.

So a very simnple dynamic dag created with a for loop, generating five dags with the same id as the source.

Three tasks, extract, transform (add som columns), and load (pandas df.to_sql) into a separate table for each source

Everything comes up as expected, five dags generated and in paused mode.

triggering one DAG works as expected, data is extracted, transformed, and then pushed into the database table. So far so good.

But, if i put more that one of the DAGs in Active, only one of the DAGs actualy update the table(s). Everything looks ok in the logs, I can see database activity, and I started to scratch my head. After a few hours of scratching my head I looked at the database side, and I can se that only one of the DAGs actually commits, and the others do an rollback!

This made me try various things, like setting autocommit to true, etc, but to no avail.

I broke out the DAG int five, just removing the for loop, and they all work as they should when active and running.

Anyone have an idea what is going on? Running Airflow 2.6.3 in Docker.

Shorted down version of my DAG:

for dyn_dag in json.loads(Variable.get('csv_data')):
    file = dyn_dag['path']
    schedule = dyn_dag['interval']
    dest_connection_id = dyn_dag['destination']
    mode = dyn_dag['mode']
    indexes = dyn_dag['indexes']
    @dag(
        schedule=schedule,
        dag_id=f"data__{table}",
        start_date=pendulum.datetime(2023, 1, 1, tz="Europe/Stockholm"),
        catchup=False,
        tags=["test"],
        doc_md = """An dynamically generated DAG"""
    )
    
    def dummy_data():

        @task()
        def extract() -> pd.DataFrame:
            df = pd.read_csv(file)
            return df      
        
        @task
        def transform(df: pd.Dataframe) -> pd.DataFrame: 

			##Add columns here removed for visibility
			
            return df
        
        @task
        def load(df: pd.DataFrame):  

            df.to_sql(file.split('.')[0], conn=dest_connection_id, if_exists=mode, index=False)
               
        load(transform(extract()))

    dummy_data()```

 

[potiuk]

I have not used Pandas but it looks like you need to manually commit your data.

Not sre what your "destination" field is in your CSV but Panda's docs is clear that if you are using SQLAlchemy connection engine the data will not be committed :

https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.to_sql.html

Using SQLAlchemy makes it possible to use any DB supported by that library. Legacy support is provided for sqlite3.Connection objects. The user is responsible for engine disposal and connection closure for the SQLAlchemy connectable. See [here](https://docs.sqlalchemy.org/en/20/core/connections.html). If passing a sqlalchemy.engine.Connection which is already in a transaction, the transaction will not be committed. If passing a sqlite3.Connection, it will not be possible to roll back the record insertion.

Likely whatever you pass there as "conn" parameter, results with a DB engine connection that is already "in-transaction" - maybe due to caching. (Wild guessing). What I'd do is to explicitly create the engine, get connection from it, pass it to the to_sql as parameter and then do connection.commit().

[ziggekatten]

I spent whole evening yesterday trying this instead of using Postgres hooks, as I suspected the same, and tried the various possibilities both with and without context manager, trying connect(), begin() as well as bind(). Still, same problem. I have some ideas still left on the table, nut was hoping for an "you stupid, you should do it like this......"

:-)

[Taragolis]

@ziggekatten That is not related to your problem, but after you resolve current one, you might have new one

for dyn_dag in json.loads(Variable.get('csv_data'))

This is not recommended way for create Dynamic DAGs, those links might help you

• Airflow Variables
  • enabling the experimental cache, 2.7.0+ (experimental, so it could be removed in any moment)
• Dynamic DAG Generation
  • Optimizing DAG parsing delays during execution

[ziggekatten]

Yep, I know that this should be avoided. This is an intermediate solution, until we have the external source provided via git/Azure Devops.