¿Alguien ha usado el sns_publish_operator?
https://airflow.apache.org/docs/stable/_modules/airflow/contrib/operators/sns_publish_operator.html
Soy bastante nuevo en el flujo de aire y tengo algunos problemas para configurar la arquitectura correctamente.
Configuré un DAG simple con una tarea de verificación de calidad de datos. Básicamente, si el conjunto de datos falla en las comprobaciones de calidad de datos, me gustaría enviar una notificación de SNS. Si pasa las comprobaciones de calidad de los datos, me gustaría que se replanteara el envío de un correo electrónico.
No parece haber tanta ayuda en línea en este ámbito como pensaba. Cualquier recurso o consejo general sería muy apreciado.
Esta pregunta es un poco más antigua, pero tal vez esto todavía ayude a alguien.
Primero dirigiéndose a SnsPublishOperator: deberá configurar una conexión Airflow a AWS. Hay varias formas de hacerlo. Lo más fácil es probablemente usar la interfaz de usuario web. Vaya allí a Admin->Conexiones->[+] (agregar nuevo registro). Luego lo configura como tipo de conexión 'Amazon Webservices Connection'. El inicio de sesión y la contraseña son la clave y el secreto de la clave de AWS. Finalmente, también debe proporcionar la región en la que se encuentra su tema SNS en la sección 'Extras':
{"region_name": "us-east-1"}Ahora, puede usar el Operador en su código, proporcionándole también el conn_id de la nueva conexión:
my_sns_task = SnsPublishOperator( task_id='task_name', target_arn='your_sns_topic_arn', message='your_message', aws_conn_id=conn_id )Teóricamente, ese operador también tiene un parámetro de "sujeto", pero recibo un error del componente cuando intento configurarlo.
En cuanto a la incorporación de ese operador en su DAG, una forma posible sería hacer que la tarea, que evalúa sus datos, falle, si las comprobaciones no pasan, y que la tarea sns se active en caso de falla:
my_sns_task = SnsPublishOperator( task_id='task_name', target_arn='your_sns_topic_arn', message='your_message', aws_conn_id=conn_id, trigger_rule='one_failed' ) my_sns_task.set_upstream(datacheck_task)