Cómo activar una tubería de CDAP utilizando operadores de airflow?

Answer

1. 

   davy.ai

   August 2, 2023 at 9:07 pm

   Para activar un pipeline de CDAP Data Fusion utilizando operadores de Airflow, puedes utilizar el HttpOperator en combinación con la API REST de CDAP disponible para cada espacio de nombres.

   Aquí tienes los pasos a seguir:

   1. Asegúrate de tener la configuración de autenticación para acceder a la API REST de CDAP. Puedes utilizar GoogleCredentials.get_application_default() para obtener las credenciales de autenticación para la cuenta de servicio predeterminada de Airflow. Asegúrate de otorgar los permisos necesarios para que la cuenta de servicio pueda acceder al espacio de nombres.
   2. Crea un HttpOperator con los parámetros necesarios para llamar a la API REST de CDAP y comenzar el pipeline. Por ejemplo:

   from airflow.operators.http_operator import SimpleHttpOperator
   
   trigger_pipeline = SimpleHttpOperator(
       task_id='trigger_pipeline',
       method='POST',
       endpoint='http://{cdap_instance}/v3/namespaces/{namespace}/apps/{application}/workflows/{workflow}/start',
       headers={'Authorization': 'Bearer {{access_token}}'},
       http_conn_id='cdap_rest_api_conn', # ID de conexión para la instancia de CDAP
       data=json.dumps({'programArguments':['arg1', 'arg2']}), # Argumentos del pipeline
       response_check=lambda response: True # Función opcional para comprobar la respuesta
   )
   

   Aquí, {cdap_instance}, {namespace}, {application} y {workflow} deberán ser reemplazados por sus valores correspondientes en el entorno. El http_conn_id deberá ser configurado con el ID de conexión para la API REST de CDAP. El parámetro data puede utilizarse para especificar cualquier argumento necesario para el trigger del pipeline.

   1. Configura la execution_date para la tarea y añádela al DAG.

   trigger_pipeline.execution_date = {{ ds }}
   

   Esto debería activar el pipeline de CDAP Data Fusion utilizando los operadores de Airflow.

Comments are closed.