python - ¿Cómo evitar que el flujo de aire de relleno dag ejecute?

scheduled-tasks airflow

Question

Supongamos que tiene un DAG de flujo de aire que no tiene sentido rellenar, lo que significa que, una vez que se haya ejecutado una vez, ejecutarlo las veces siguientes rápidamente sería completamente inútil.

Por ejemplo, si está cargando datos de una fuente que solo se actualiza cada hora en su base de datos, el relleno, que se produce en una sucesión rápida, simplemente estaría importando los mismos datos una y otra vez.

Esto es especialmente molesto cuando crea una instancia de una nueva tarea por hora y se ejecuta N cantidad de veces por cada hora que perdió, realizando un trabajo redundante, antes de que comience a ejecutarse en el intervalo que especificó.

La única solución que se me ocurre es algo que se desaconseja específicamente en las Preguntas frecuentes de los documentos.

Recomendamos no usar valores dinámicos como start_date, especialmente datetime.now() ya que puede ser bastante confuso.

¿Hay alguna forma de deshabilitar el relleno de un DAG o debo hacer lo anterior?

Answer 1

Actualice a la versión 1.8 del flujo de aire y use catchup_by_default = False en airflow.cfg o aplique catchup = False en cada uno de sus dags.

https://github.com/apache/incubator-airflow/blob/master/UPDATING.md#catchup_by_default

Answer 2

Esto parece ser un problema de flujo de aire no resuelto. Sé que realmente me gustaría tener exactamente la misma característica. Aquí está hasta donde he llegado; puede ser útil para otros.

Son características de la interfaz de usuario (al menos en 1.7.1.3) que pueden ayudar con este problema. Si va a la vista en árbol y hace clic en una tarea específica (recuadros cuadrados), aparecerá un botón de diálogo con el botón "marcar éxito". Al hacer clic en ''pasado'', y luego en ''marcar éxito'' se etiquetarán todas las instancias de esa tarea en DAG como exitosas y no se ejecutarán. El DAG de nivel superior (círculos en la parte superior) también se puede etiquetar como exitoso de una manera similar, pero no parece haber forma de etiquetar varias instancias de DAG.

Todavía no lo he investigado con suficiente profundidad, pero puede ser posible usar el subcomando ''trigger_dag'' para marcar los estados de los DAG. vea aquí: https://github.com/apache/incubator-airflow/pull/644/commits/4d30d4d79f1a18b071b585500474248e5f46d67d

Una característica de CLI para marcar DAG está en proceso: http://mail-archives.apache.org/mod_mbox/airflow-commits/201606.mbox/%[email protected]%3E https://github.com/apache/incubator-airflow/pull/1590

ACTUALIZACIÓN (28/09/2016): Se ha agregado un nuevo operador ''LatestOnlyOperator'' ( https://github.com/apache/incubator-airflow/pull/1752 ) que solo ejecutará la última versión de tareas posteriores. Suena muy útil y espero que pronto llegue a los lanzamientos

ACTUALIZACIÓN 2: A partir del flujo de aire 1.8, se ha lanzado LatestOnlyOperator .