Airflow 1.9 如何将任务日志发送至Elasticsearch？

嘿，我来帮你搞定Airflow 1.9把任务日志发送到Elasticsearch的配置问题！下面是一步步的实操指南：

1. 先安装必要依赖 #

Airflow 1.9的Elasticsearch日志处理器需要依赖官方的Elasticsearch Python库，而且要注意版本兼容（1.9对ES的适配更偏向5.x/6.x版本，不建议用7.x及以上）。执行以下命令安装：

pip install elasticsearch>=5.0.0,<6.0.0

2. 修改 Airflow 配置文件 (airflow.cfg) #

找到你的Airflow主配置文件（通常在~/airflow/airflow.cfg或者/etc/airflow/airflow.cfg），做以下两处修改：

2.1 配置日志读取器 #

找到[core]小节，修改task_log_reader参数，指定使用Elasticsearch日志处理器：

[core]
task_log_reader = elasticsearch.task

2.2 配置Elasticsearch连接与日志格式 #

在配置文件中找到（如果没有就新增）[elasticsearch]小节，填入以下核心配置：

[elasticsearch]
host = http://localhost:9200  # 替换成你的ES集群地址
log_id_template = {{dag_id}}-{{task_id}}-{{execution_date}}-{{try_number}}
end_of_log_mark = END_OF_LOG
write_stdout = True
json_format = True
json_format_indent = 2
index_template = airflow-logs-%Y-%m-%d  # 按日期分片的索引模板，也可以用dag_id区分

解释几个关键参数：

• write_stdout：设为True会把任务的标准输出也同步到ES
• json_format：用JSON格式存储日志，方便后续在Kibana中分析
• index_template：支持模板变量，比如%Y-%m-%d会自动按日期生成索引

3. 验证你找到的配置模板适用性 #

如果仓库里的配置模板是针对Airflow 1.9版本的，那核心参数基本是通用的，但要注意：

• 新版本Airflow（比如1.10+）新增的参数（比如elasticsearch_conn_id）在1.9里是不支持的，别照搬
• 重点核对[core]下的task_log_reader和[elasticsearch]下的host、index_template这几个关键参数，确保和上面的示例匹配

4. 重启 Airflow 服务 #

配置修改后，必须重启Airflow的相关服务才能生效：

# 重启webserver
airflow webserver -D
# 重启scheduler
airflow scheduler -D
# 如果用Celery worker，也要重启worker
airflow worker -D

5. 测试与排查问题 #

运行一个简单的测试任务（比如BashOperator执行echo "test log"），然后：

1. 去Elasticsearch中查询对应的索引（比如airflow-logs-YYYY-MM-DD），看是否生成了日志文档
2. 如果没看到日志，检查：
   • ES服务是否正常运行，Airflow服务器能访问到ES地址
   • 安装的elasticsearch库版本是否符合要求
   • Airflow日志目录是否有写入权限（虽然日志是发ES，但本地还是会存一份临时日志）
   • 查看Airflow自身的日志（比如~/airflow/logs/scheduler），看是否有连接ES的报错

内容的提问来源于stack exchange，提问作者Amit Kumar