You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Mrhacktivist (via GitHub)" <gi...@apache.org> on 2024/03/26 13:38:35 UTC
[I] sql.py from the postgreshook not writing sql statements [airflow]
Mrhacktivist opened a new issue, #38498:
URL: https://github.com/apache/airflow/issues/38498
### Apache Airflow Provider(s)
postgres
### Versions of Apache Airflow Providers
5.5.1
### Apache Airflow version
2.6.3
### Operating System
amazon linux 2
### Deployment
Official Apache Airflow Helm Chart
### Deployment details
_No response_
### What happened
When using the predfined list, code is wirting the sql stamenets iin logs -
Example -
============================
from datetime import datetime
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.hooks.postgres_hook import PostgresHook
# Define the function to execute SQL script using PostgresHook
def run_stage_queries(table_name, **kwargs):
print(table_name)
print(records)
load_type_res = pg_hook.get_records(
"SELECT * FROM public.test WHERE ods_table_name = '{}'".format(table_name))
print('Executing SQL query: {}'.format(load_type_res))
print('&&&& ', load_type_res[0][0])
load_type = load_type_res[0][0]
# Define the DAG
dag = DAG(
dag_id="vaibhav",
schedule_interval="@daily",
start_date=datetime(2023, 1, 1),
max_active_runs=1,
catchup=False
)
# Fetching records within the context of the DAG
pg_hook = PostgresHook(postgres_conn_id='red')
sql_ods = "SELECT ods_table_name FROM public.test;"
records = ["vaibhav","nikhil","kunal"]
ods_table_list = []
for o_table in records:
if o_table is not None:
ods_table_list.append(o_table)
if len(ods_table_list) > 0:
init_parameters = DummyOperator(task_id='Load_base_tables', dag=dag)
stage_Load_Success = DummyOperator(task_id='Stage_Load_Success', dag=dag)
for tableName in ods_table_list:
load_base_stg = PythonOperator(
task_id=tableName + '_stage',
op_kwargs={'table_name': tableName},
python_callable=run_stage_queries,
provide_context=True,
dag=dag)
init_parameters >> load_base_stg
load_base_stg >> stage_Load_Success
Output -
[2024-03-26, 12:15:36 UTC] {{logging_mixin.py:150}} INFO - ['vaibhav', 'nikhil', 'kunal']
[2024-03-26, 12:15:37 UTC] {{sql.py:374}} INFO - Running statement: SELECT * FROM public.test WHERE ods_table_name = 'vaibhav', parameters: None
[2024-03-26, 12:15:37 UTC] {{sql.py:383}} INFO - Rows affected: 1
[2024-03-26, 12:15:37 UTC] {{logging_mixin.py:150}} INFO - Executing SQL query: [('vaibhav',)]
[2024-03-26, 12:15:37 UTC] {{logging_mixin.py:150}} INFO - &&&& vaibhav
[2024-03-26, 12:15:37 UTC] {{python.py:183}} INFO - Done. Returned value was: None
[2024-03-26, 12:15:37 UTC] {{taskinstance.py:1345}} INFO - Marking task as SUCCESS. dag_id=vaibhav, task_id=vaibhav_stage, execution_date=20240326T121522, start_date=20240326T121536, end_date=20240326T121537
[2024-03-26, 12:15:37 UTC] {{local_task_job_runner.py:225}} INFO - Task exited with return code 0
[2024-03-26, 12:15:37 UTC] {{taskinstance.py:2653}} INFO - 0 downstream tasks scheduled from follow-on schedule check
======================
When the variable genereated by iterating through the get_records, code is not writing logs -
from datetime import datetime
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.hooks.postgres_hook import PostgresHook
# Define the function to execute SQL script using PostgresHook
def run_stage_queries(table_name, **kwargs):
print(table_name)
print(records)
print(ods_table_list)
load_type_res = pg_hook.get_records(
"SELECT * FROM public.test WHERE ods_table_name = '{}'".format(table_name))
print('Executing SQL query: {}'.format(load_type_res))
print('&&&& ', load_type_res[0][0])
load_type = load_type_res[0][0]
# Define the DAG
dag = DAG(
dag_id="vaibhavvvvv",
schedule_interval="@daily",
start_date=datetime(2023, 1, 1),
max_active_runs=1,
catchup=False
)
# Fetching records within the context of the DAG
pg_hook = PostgresHook(postgres_conn_id='red')
sql_ods = "SELECT ods_table_name FROM public.test;"
records = pg_hook.get_records(sql_ods)
vaibhavvv = [item[0] for item in records]
ods_table_list = vaibhavvv
if len(ods_table_list) > 0:
init_parameters = DummyOperator(task_id='Load_base_tables', dag=dag)
stage_Load_Success = DummyOperator(task_id='Stage_Load_Success', dag=dag)
for tableName in ods_table_list:
load_base_stg = PythonOperator(
task_id=tableName + '_stage',
op_kwargs={'table_name': tableName},
python_callable=run_stage_queries,
provide_context=True,
dag=dag)
init_parameters >> load_base_stg
load_base_stg >> stage_Load_Success
Output -
[2024-03-26, 12:16:30 UTC] {{standard_task_runner.py:85}} INFO - Job 52: Subtask vaibhav_stage
[2024-03-26, 12:16:30 UTC] {{taskinstance.py:1545}} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='vaibhavvvvv' AIRFLOW_CTX_TASK_ID='vaibhav_stage' AIRFLOW_CTX_EXECUTION_DATE='2024-03-26T12:16:14.873032+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2024-03-26T12:16:14.873032+00:00'
[2024-03-26, 12:16:30 UTC] {{logging_mixin.py:150}} INFO - vaibhav
[2024-03-26, 12:16:30 UTC] {{logging_mixin.py:150}} INFO - [('vaibhav',), ('nikhil',), ('kunal',)]
[2024-03-26, 12:16:30 UTC] {{logging_mixin.py:150}} INFO - ['vaibhav', 'nikhil', 'kunal']
[2024-03-26, 12:16:30 UTC] {{logging_mixin.py:150}} INFO - Executing SQL query: [('vaibhav',)]
[2024-03-26, 12:16:30 UTC] {{logging_mixin.py:150}} INFO - &&&& vaibhav
[2024-03-26, 12:16:30 UTC] {{python.py:183}} INFO - Done. Returned value was: None
Issue is not MWAA end, requested cx to open a issue with open source airflow.
### What you think should happen instead
_No response_
### How to reproduce
Can leverage the 2 code given above
### 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
Re: [I] sql.py imported from sql in postgreshook not writing sql statements in the task logs [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on issue #38498:
URL: https://github.com/apache/airflow/issues/38498#issuecomment-2026249290
I think someone might take a look if you spend a little time on explaining what exactly is wrog. I spend about a minute looking at your description and have no idea not even what the problem is but even what kind of problem you have (this how much my free time away of my family and other things I do a day before Easter).
If you spend asom more time and describe what problem you really have in the way that some volunteers might understand (remember that this is a place that volunteer time is spend to help people like you), that heavily increases your chances that somone will spend a few minutes of theoir time trying to help to solve a problem you have.
I heartily recommend to be very precise and state your problem clearly, that really helps to get help.
BTW. Converting it into a discussion - with the description there is no way to say if this is even a problem at all.
--
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
Re: [I] sql.py imported from sql in postgreshook not writing sql statements in the task logs [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed issue #38498: sql.py imported from sql in postgreshook not writing sql statements in the task logs
URL: https://github.com/apache/airflow/issues/38498
--
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
Re: [I] sql.py from the postgreshook not writing sql statements in the task logs [airflow]
Posted by "Mrhacktivist (via GitHub)" <gi...@apache.org>.
Mrhacktivist commented on issue #38498:
URL: https://github.com/apache/airflow/issues/38498#issuecomment-2025630347
Can someone please take a look
--
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