为气流中的日志设置s3

Question

我正在使用docker-compose来设置可扩展的气流群集.我的解决方案基于这个Dockerfile https://hub.docker.com/r/puckel/docker-airflow/

我的问题是将日志设置为从s3写入/读取.当一个dag完成后,我得到这样的错误

*** Log file isn't local.
*** Fetching here: http://ea43d4d49f35:8793/log/xxxxxxx/2017-06-26T11:00:00
*** Failed to fetch log file from worker.

*** Reading remote logs...
Could not read logs from s3://buckets/xxxxxxx/airflow/logs/xxxxxxx/2017-06-
26T11:00:00

我在这个airflow.cfg文件中设置了一个新的部分

[MyS3Conn]
aws_access_key_id = xxxxxxx
aws_secret_access_key = xxxxxxx
aws_default_region = xxxxxxx

然后在远程日志部分中指定s3路径 airflow.cfg

remote_base_log_folder = s3://buckets/xxxx/airflow/logs
remote_log_conn_id = MyS3Conn

我是否正确设置了这个并且有错误？这里有成功的秘诀吗？

- 更新

我尝试以URI和JSON格式导出,似乎都不起作用.然后我导出了aws_access_key_id和aws_secret_access_key,然后气流开始捡起它.现在我在工作日志中得到了他的错误

6/30/2017 6:05:59 PMINFO:root:Using connection to: s3
6/30/2017 6:06:00 PMERROR:root:Could not read logs from s3://buckets/xxxxxx/airflow/logs/xxxxx/2017-06-30T23:45:00
6/30/2017 6:06:00 PMERROR:root:Could not write logs to s3://buckets/xxxxxx/airflow/logs/xxxxx/2017-06-30T23:45:00
6/30/2017 6:06:00 PMLogging into: /usr/local/airflow/logs/xxxxx/2017-06-30T23:45:00

- 更新

我也找到了这个链接 https://www.mail-archive.com/dev@airflow.incubator.apache.org/msg00462.html

然后我进入我的一个工作机器(与Web服务器和调度程序分开)并在python中运行这段代码

import airflow
s3 = airflow.hooks.S3Hook('s3_conn')
s3.load_string('test', airflow.conf.get('core', 'remote_base_log_folder'))

我收到此错误.

boto.exception.S3ResponseError: S3ResponseError: 403 Forbidden

我尝试导出几种不同类型的AIRFLOW_CONN_env,如https://airflow.incubator.apache.org/concepts.html连接部分和此问题的其他答案所述.

s3://<AWS_ACCESS_KEY_ID>:<AWS_SECRET_ACCESS_KEY>@S3

{"aws_account_id":"<xxxxx>","role_arn":"arn:aws:iam::<xxxx>:role/<xxxxx>"}

{"aws_access_key_id":"<xxxxx>","aws_secret_access_key":"<xxxxx>"}

我还导出了AWS_ACCESS_KEY_ID和AWS_SECRET_ACCESS_KEY但没有成功.

这些凭据存储在数据库中,因此一旦我在UI中添加它们,它们就应该被工作人员选中,但由于某些原因它们无法写入/读取日志.

Answer 1

更新Airflow 1.10使记录更容易.

对于s3日志记录,请根据上述答案设置连接挂钩

然后只需将以下内容添加到airflow.cfg中

    [core]
    # Airflow can store logs remotely in AWS S3. Users must supply a remote
    # location URL (starting with either 's3://...') and an Airflow connection
    # id that provides access to the storage location.
    remote_base_log_folder = s3://my-bucket/path/to/logs
    remote_log_conn_id = MyS3Conn
    # Use server-side encryption for logs stored in S3
    encrypt_s3_logs = False

对于gcs日志记录,

1. 首先安装gcp_api软件包,如下所示:pip install apache-airflow [gcp_api].

2. 根据上面的答案设置连接挂钩

3. 将以下内容添加到airflow.cfg中

   [core]
   # Airflow can store logs remotely in AWS S3. Users must supply a remote
   # location URL (starting with either 's3://...') and an Airflow connection
   # id that provides access to the storage location.
   remote_logging = True
   remote_base_log_folder = gs://my-bucket/path/to/logs
   remote_log_conn_id = MyGCSConn
   

注意:从Airflow 1.9开始,远程日志记录已经发生了重大变化.如果您使用的是1.9,请继续阅读.

参考这里

完整说明:

1. 创建一个存储配置的目录并放置它,以便在PYTHONPATH中找到它.一个例子是$ AIRFLOW_HOME/config

2. 创建名为$ AIRFLOW_HOME/config/log_config.py和$ AIRFLOW_HOME/config/__ init__.py的空文件

3. 将airflow/config_templates/airflow_local_settings.py的内容复制到刚刚在上面的步骤中创建的log_config.py文件中.

4. 自定义模板的以下部分:

   #Add this variable to the top of the file. Note the trailing slash.
   S3_LOG_FOLDER = 's3://<bucket where logs should be persisted>/'
   
   Rename DEFAULT_LOGGING_CONFIG to LOGGING CONFIG
   LOGGING_CONFIG = ...
   
   Add a S3TaskHandler to the 'handlers' block of the LOGGING_CONFIG variable
   's3.task': {
       'class': 'airflow.utils.log.s3_task_handler.S3TaskHandler',
       'formatter': 'airflow.task',
       'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
       's3_log_folder': S3_LOG_FOLDER,
       'filename_template': FILENAME_TEMPLATE,
   },
   
    Update the airflow.task and airflow.task_runner blocks to be 's3.task' instead >of 'file.task'.
   'loggers': {
       'airflow.task': {
           'handlers': ['s3.task'],
           ...
       },
       'airflow.task_runner': {
           'handlers': ['s3.task'],
           ...
       },
       'airflow': {
           'handlers': ['console'],
           ...
       },
   }
   

5. 根据上述答案,确保在Airflow中定义了s3连接挂钩.钩子应具有对S3_LOG_FOLDER中上面定义的s3桶的读写访问权限.

6. 更新$ AIRFLOW_HOME/airflow.cfg以包含:

   task_log_reader = s3.task
   logging_config_class = log_config.LOGGING_CONFIG
   remote_log_conn_id = <name of the s3 platform hook>
   

7. 重新启动Airflow Web服务器和调度程序,并触发(或等待)新任务执行.

8. 验证日志是否显示在您定义的存储桶中新执行的任务中.

9. 验证s3存储查看器是否在UI中正常工作.拉出新执行的任务,并验证您是否看到类似的内容:

   *** Reading remote log from gs://<bucket where logs should be persisted>/example_bash_operator/run_this_last/2017-10-03T00:00:00/16.log.
   [2017-10-03 21:57:50,056] {cli.py:377} INFO - Running on host chrisr-00532
   [2017-10-03 21:57:50,093] {base_task_runner.py:115} INFO - Running: ['bash', '-c', u'airflow run example_bash_operator run_this_last 2017-10-03T00:00:00 --job_id 47 --raw -sd DAGS_FOLDER/example_dags/example_bash_operator.py']
   [2017-10-03 21:57:51,264] {base_task_runner.py:98} INFO - Subtask: [2017-10-03 21:57:51,263] {__init__.py:45} INFO - Using executor SequentialExecutor
   [2017-10-03 21:57:51,306] {base_task_runner.py:98} INFO - Subtask: [2017-10-03 21:57:51,306] {models.py:186} INFO - Filling up the DagBag from /airflow/dags/example_dags/example_bash_operator.py
   

Answer 2

您需要通过气流UI设置s3连接.为此,您需要转到气流UI上的Admin - > Connections选项卡,并为S3连接创建一个新行.

一个示例配置是:

Conn Id:my_conn_S3

Conn类型:S3

额外:{"aws_access_key_id":"your_aws_key_id","aws_secret_access_key":"your_aws_secret_key"}

Answer 3

(自Airflow 1.10.2起更新)

如果您不使用管理界面,这是一个解决方案.

我的Airflow不能在持久性服务器上运行...(它每天都会在Heroku上的Docker容器中重新启动.)我知道我错过了很多很棒的功能,但在我的最小设置中,我永远不要触摸管理界面或cfg文件.相反,我必须在bash脚本中设置特定于Airflow的环境变量,该脚本会覆盖.cfg文件.

Apache的气流[S3]

首先,您需要s3安装子包才能将Airflow日志写入S3.(boto3适用于DAG中的Python作业,但S3Hook取决于s3子包.)

还有一个注意事项:conda install 还没有处理这个,所以我必须这样做pip install apache-airflow[s3].

环境变量

在bash脚本中,我设置了这些core变量.从这些说明开始,但使用AIRFLOW__{SECTION}__{KEY}环境变量的命名约定,我做:

export AIRFLOW__CORE__REMOTE_LOGGING=True
export AIRFLOW__CORE__REMOTE_BASE_LOG_FOLDER=s3://bucket/key
export AIRFLOW__CORE__REMOTE_LOG_CONN_ID=s3_uri
export AIRFLOW__CORE__ENCRYPT_S3_LOGS=False

S3连接ID

的s3_uri上面是,我由一个连接ID.在Airflow中,它对应于另一个环境变量AIRFLOW_CONN_S3_URI.它的值是您的S3路径,必须采用URI形式.那是

s3://access_key:secret_key@bucket/key

存储此,但您处理其他敏感的环境变量.

通过此配置,Airflow将能够将您的日志写入S3.他们将遵循的道路s3://bucket/key/dag/task_id/timestamp/1.log.

有关从Airflow 1.8升级到Airflow 1.10的附录

我最近将我的生产管道从Airflow 1.8升级到1.9,然后升级到1.10.好消息是变化很小; 其余的工作只是找出包安装的细微差别(与原始问题有关S3日志无关).

(1)首先,我需要使用Airflow 1.9升级到Python 3.6.

(2)包名称从改变airflow到apache-airflow1.9.你也可能会遇到这种情况pip install.

(3)包装psutil必须在Airflow的特定版本范围内.你可能会遇到这种情况pip install apache-airflow.

(4)Airflow 1.9+需要python3-dev头文件.

(5)以下是实质性变化:export AIRFLOW__CORE__REMOTE_LOGGING=True现在是必需的.和

(6)S3中的日志路径略有不同,我在答案中更新了:s3://bucket/key/dag/task_id/timestamp/1.log.

但就是这样!日志在1.9中不起作用,所以我建议直接使用1.10,现在可以使用了.