Airflow y superQuery

Monitorea en tiempo real tus costos de BigQuery con el SuperQueryOperator

Foto de Soheb Zaidi en Unsplash

"¿Cuánto cuesta?" Una pregunta tan habitual en el mundo tech que cualquier persona en una start-up se estremece al escucharla. Y la respuesta, casi siempre, es: "No estamos seguros".

En el mundo de la ingeniería de datos, una de las mejores herramientas para orquestar workflows es Apache Airflow. Esta herramienta ha sacado a muchas empresas del estancamiento de los rígidos cron schedulers para llevarlas a surfear las olas del big data en mar abierto, gracias a los Directed Acyclic Graphs (DAGs).

Claro que esto implica mover grandes volúmenes de datos hacia y desde las bases de datos, y con ese movimiento suelen llegar costos inevitables.

Una de esas bases de datos —una supercomputadora, si quieres llamarla así— es Google BigQuery. Es el producto estrella de Google Cloud y permite procesar datos a escala de Petabytes. Te permite preocuparte menos por la potencia de la infraestructura y más por la calidad de tus análisis y por los problemas de flujo de datos que necesitas resolver.

Un factor clave a considerar con BigQuery es qué tan fácil resulta, para cualquier persona u organización, disparar los costos al escanear datos en la plataforma. Incluso los data engineers más experimentados te contarán con angustia esos errores en los que terminaron escaneando datos que en realidad no necesitaban y dispararon la factura mensual de análisis muy por encima del presupuesto.

Aquí entra superQuery. Su filosofía es que no tengas que preocuparte por los costos, porque dispones de toda la información necesaria y de algunos resguardos que te permiten tomar una decisión informada.

Conoce tus costos en Airflow

Mientras tus DAGs de Airflow trabajan tranquilamente y envían datos a tus sistemas de procesamiento preferidos, en segundo plano se acumula una gran cantidad de logs. Los logs de Airflow son fáciles de consultar, simples de leer y te dan una buena visión general de lo que hace tu DAG. ¿No sería genial que esos archivos también te mostraran información sobre el plan de ejecución de tu query y, en concreto, cuál fue el costo y el total de datos escaneados? ¡Claro que sí! Algo así:

--------------------------------------------------------------------
Starting attempt 1 of 4
--------------------------------------------------------------------

[2019-03-11 21:12:02,129] {models.py:1593} INFO - Executing <Task(SuperQueryOperator): connect_to_superquery_proxy> on 2019-03-01T00:00:00+00:00
[2019-03-11 21:12:03,836] {superq_operators.py:54} INFO - Executing: #standardSQL
SELECT COUNT(testField) FROM `mydata.PROD.myTable`;
[2019-03-11 21:12:03,844] {logging_mixin.py:95} INFO - [2019-03-11 21:12:03,843] {base_hook.py:83} INFO - Using connection to: id: mysql_default. Host: superproxy.system.io, Port: 3306, Schema: None, Login: XXXXXX, Password: XXXXXXXX, extra: {}
[2019-03-11 21:12:15,172] {superq_operators.py:68} INFO - ((
'{
   "startTime":1552331525642,
   "endTime":1552331534624,
   "executionTime":"8988",
   "bigQueryTotalBytesProcessed":26388279066,
   "bigQueryTotalCost":"0.12",
   "superQueryTotalBytesProcessed":0,
   "superQueryTotalCost":"0.00",
   "saving":0,
   "totalRows":"1",
}', '', '1', 'true'),)
[2019-03-11 21:12:17,121] {logging_mixin.py:95} INFO - [2019-03-11 21:12:17,119] {jobs.py:2527} INFO - Task exited with return code 0

Este fragmento del log te indica que tu operador de BigQuery en Airflow escaneó 24Gb de datos y te costó $0.12. Así de simple. También puedes parsear los archivos de log con una herramienta de terceros o con un script bash y armar un resumen de los costos generados por los DAGs que escanean datos de BigQuery.

¿Cómo funciona todo esto?

SuperQuery utiliza un proxy de MySql para universalizar la conectividad y ofrecer una interfaz SQL desde la cual obtener información.

Lo que necesitas a continuación: el SuperQueryOperator

Estos son los pasos para obtener la misma funcionalidad descrita arriba:

1. Agrega el plugin de superquery a Airflow para poder usar el SuperQueryOperator.
2. Suscríbete a la prueba de superQuery y obtén las credenciales de acceso al proxy de MySql de superQuery.
3. Prueba tu conexión al proxy con el DAG que aparece más abajo.
4. Reemplaza tus operadores de BigQuery por el SuperQuery operator en tus propios DAGs cuando quieras usar esta funcionalidad.

Cómo usar el SuperQueryOperator

Esta es la interfaz del operador de SuperQuery:

TEST_SQL = """#standardSQL

SELECT COUNT(*) FROM `mydata.PROD.myTable`;"""

SuperQueryOperator(

  task_id="connect_to_superquery_proxy",

  sql=TEST_SQL,

  database="",

  explain=True,  # False if you don't want information

  dag=dag

)

Este es el código del operador, que debes copiar en la carpeta plugins de Airflow:

https://gist.github.com/super-eben/b1d49538b46fa3f55c5bddfe73405b34

Y aquí tienes código para probar tu conexión a SuperQuery:

https://gist.github.com/super-eben/dcfa4420c419331e5a12a8b23e0a088c

En resumen

Este post explica cómo obtener visibilidad de tus costos de Airflow al conectar e implementar tareas hacia y desde BigQuery. El uso del proxy de SuperQuery se puede extender para incluir una amplia variedad de información detallada del plan de ejecución y aprovechar al máximo los beneficios que ofrece el sistema.

¡Feliz monitoreo (y ahorro) de costos!