Ytukyg

I have a dag as below:
ingest_excel.py:

from __future__ import print_function



import time

from builtins import range

from datetime import timedelta

from pprint import pprint



import airflow

from airflow.models import DAG

#from airflow.operators.bash_operator import BashOperator

from airflow.operators.python_operator import PythonOperator



args = {

    'owner': 'rxie',

    'start_date': airflow.utils.dates.days_ago(2),

}



dag = DAG(

    dag_id='ingest_excel',

    default_args=args,

    schedule_interval='0 0 * * *',

    dagrun_timeout=timedelta(minutes=60),

)



def print_context(**kwargs):

    pprint("DAG info below:")

    pprint(kwargs)

    return 'Whatever you return gets printed in the logs'





t11_extract_excel_to_csv = PythonOperator(

    task_id='t1_extract_excel_to_csv',

    provide_context=True,

    python_callable=print_context(),

    op_kwargs=None,

    dag=dag,

)





t12_upload_csv_to_hdfs_parquet = PythonOperator(

    task_id='t12_upload_csv_to_hdfs_parquet',

    provide_context=True,

    python_callable=print_context(),

    op_kwargs=None,

    dag=dag,

)





t13_register_parquet_to_impala = PythonOperator(

    task_id='t13_register_parquet_to_impala',

    provide_context=True,

    python_callable=print_context(),

    op_kwargs=None,

    dag=dag,

)



t21_text_to_parquet = PythonOperator(

    task_id='t21_text_to_parquet',

    provide_context=True,

    python_callable=print_context(),

    op_kwargs=None,

    dag=dag,

)



t22_register_parquet_to_impala = PythonOperator(

    task_id='t22_register_parquet_to_impala',

    provide_context=True,

    python_callable=print_context(),

    op_kwargs=None,

    dag=dag,

)



t31_verify_completion = PythonOperator(

    task_id='t31_verify_completion',

    provide_context=True,

    python_callable=print_context(),

    op_kwargs=None,

    dag=dag,

)



t32_send_notification = PythonOperator(

    task_id='t32_send_notification',

    provide_context=True,

    python_callable=print_context(),

    op_kwargs=None,

    dag=dag,

)



t11_extract_excel_to_csv >> t12_upload_csv_to_hdfs_parquet

t12_upload_csv_to_hdfs_parquet >> t13_register_parquet_to_impala



t21_text_to_parquet >> t22_register_parquet_to_impala





t13_register_parquet_to_impala >> t31_verify_completion

t22_register_parquet_to_impala >> t31_verify_completion



t31_verify_completion >> t32_send_notification





#if __name__ == "__main__":

#    dag.cli()

In DAG GUI it prompts:

Broken DAG: [/root/airflow/dags/ingest_excel.py] python_callable
param must be callable

This is my first dag in Airflow, and I am pretty new to Airflow, it would be greatly appreciated if anyone can shed me some light and sort it out for me.

Thank you in advance.

To elaborate on your issue: your process is broken because you're not passing the function print_context to the PythonOperator, you're passing the result of calling print_context:

[...]



t32_send_notification = PythonOperator(

    task_id='t32_send_notification',

    provide_context=True,

    python_callable=print_context(), # <-- This is the issue.

    op_kwargs=None,

    dag=dag,

)



[...]

Your function is returning the string 'Whatever you return gets printed in the logs' which is, in turn, being provided to the PythonOperator in the python_callable keyword argument. Airflow is essentially attempting to do the following:

your_return = 'Whatever you return gets printed in the logs'

your_return()

...and you're receiving the error you see. The other contributor is correct in stating that you should change your PythonOperator.python_callable keyword argument to simply print_context

I'm not entirely sure why you're code doesn't work. It should work, but a work around is given below.

def print_context(**kwargs):

ds = kwargs['ds']

also the python_callable should be passed like this

python_callable=print_context,