AirFlow 日常任务的跑批，常常伴随着大量日志文件的产生，尤其是在任务失败尝试重启的时候，日志文件会变得异常地大。这些文件如果不能及时清理，很可能导致服务器磁盘拉爆，下面我们编写一个每天自动清理日志的任务，只保留近一个月的日志。

同时，MySQL 执行日志也应该定期进行清理，不然 MySQL 中 Airflow 库占用空间会越来越大，放任不管的话，迟早有一天 MySQL 库会崩掉。

1. 日志占用磁盘大小

先来看一下运行日志会占用我们磁盘多大的空间，可以看到轻轻松松就会占用 140G 数据占用了，这还只是一个月的数据，所以每天清理日志数据是非常有必要的。

  1. $ du -sh ~/bigdata/airflow/logs
  2. 140G    logs/

2. 清理本地调度文件

AirFlow 任务运行的日志文件分为两部分，一部分为本地日志文件，它保存在 $AIRFLOW_HOME/logs/scheduler 文件夹下，通常你查看运行日志时的数字，浏览器就会下载一份文件（这份文件便是存储在 logs/scheduler 文件夹下的）。
image.png

实现的代码也很简单，就是使用 BashOperator 执行一段删除代码即可，这里设置为只保存近 30 天的日志文件，其余的都删掉。

  1. from airflow.operators.bash_operator import BashOperator
  4. airflow_home = os.environ.get("AIRFLOW_HOME")
  5. logging_dir = os.path.join(airflow_home, "logs", "scheduler")
  6. clean_templated = """rm -rf %s/{{ macros.ds_add(ds, -30) }}""" % logging_dir
  8. cleanup_op = BashOperator(
  9.     task_id="cleanup_logging",
  10.     bash_command=clean_templated,
  11.     dag=dag)

3. 清理MySQL库中的执行日志

借助于 MySqlOperator 来清理 MySQL 中 Airflow 执行日志，前提是 Airflow 的数据库使用的是 MySQL。

⭐ 值得注意的是，如果有任务的执行周期是 1 个月或者 2 个月执行一次，需要修改 ds_add(ds, -7)

  1. from airflow.providers.mysql.operators.mysql import MySqlOperator
  4. tk = {
  5.     "xcom":"execution_date",
  6.     "task_instance":"execution_date",
  7.     "task_fail":"execution_date",
  8.     "sla_miss":"execution_date",
  9.     "log":"execution_date",
  10.     "job":"start_date",
  11.     "dag_run":"execution_date"
  12.     }
  14. for tb in tk:
  15.     del_op = MySqlOperator(
  16.         task_id=f"del_{tb}",
  17.         sql="DELETE FROM %s WHERE %s < '{{ ds_add(ds, -7) }}'" % (tb, tk[tb]),
  18.         dag=dag,
  19.         )

3. 完整脚本

以下为完整的清理脚本

  1. import os
  2. from airflow import DAG
  3. from datetime import timedelta
  4. from airflow.utils.dates import days_ago
  5. from airflow.operators.bash_operator import BashOperator
  6. from airflow.providers.mysql.operators.mysql import MySqlOperator
  8. airflow_home = os.environ.get("AIRFLOW_HOME")
  9. logging_dir = os.path.join(airflow_home, "logs", "scheduler")
  10. clean_templated = "rm -rf %s/{{ macros.ds_add(ds, -30) }}" % logging_dir
  12. __doc__ = "自动清理DAG日志"
  15. default_args = {
  16.     'owner': 'yumingmin',
  17.     'depends_on_past': False,
  18.     'email': ['yu_mingm623@163.com'],
  19.     'email_on_failure': False,
  20.     'email_on_retry': False,
  21.     'retries': 1,
  22.     'retry_delay': timedelta(minutes=30),
  23.     'catchup': False
  24.     }
  26. dag = DAG(
  27.     'cleanup_logging',
  28.     default_args=default_args,
  29.     description='Cleanup logging automatically',
  30.     schedule_interval=timedelta(days=1),
  31.     start_date=days_ago(1),
  32.     tags=['prod'],
  33.     )
  35. cleanup_op = BashOperator(
  36.     task_id="cleanup_logging",
  37.     bash_command=clean_templated,
  38.     dag=dag)
  40. tk = {
  41.     "xcom":"execution_date",
  42.     "task_instance":"execution_date",
  43.     "task_fail":"execution_date",
  44.     "sla_miss":"execution_date",
  45.     "log":"execution_date",
  46.     "job":"start_date",
  47.     "dag_run":"execution_date"
  48.     }
  50. for tb in tk:
  51.     del_op = MySqlOperator(
  52.         task_id=f"del_{tb}",
  53.         sql="DELETE FROM %s WHERE %s < '{{ ds_add(ds, -30) }}'" % (tb, tk[tb]),
  54.         dag=dag,
  55.         )
  56.     cleanup_op >> del_op
  58. # Documentation for Dag and Task
  59. dag.doc_md = dedent(
  60.     """
  61.     # 自动清理DAG日志
  63.     - 仅仅保留<font color='red'>30</font>天内的运行日志，其余都会被删除
  64.     - 脚本使用 Bash 作为媒介来清除 logs 下 schedule 相关日期文件夹
  65.     """
  66.     )

，开启该任务就会每天清理前 31 天