如何从 DataProcPySparkOperator 发出气流日志

Question

我已经看到了一些关于从 PythonOperator 和 configuring Airflow logs 发出日志的信息，但还没有发现任何可以让我从容器化进程中发出日志的东西，例如DataProcPySparkOperator。

我已经在 Operator 集群中运行的 pyspark 脚本的顶部添加了以下内容：

import logging
logging.info('Test bare logger')
for ls in ['airflow', 'airflow.task', __name__]:
    l = logging.getLogger(ls)
    l.info('Test {} logger'.format(ls))
print('Test print() logging')

它不会产生任何输出，尽管 Operator 脚本会按预期运行。

我假设我可以从集群内建立到云存储（或数据库）的连接，也许可以捎带用于读取和写入文件的现有连接，但是......这似乎就像为了一个共同的需要做很多工作。我非常希望在计算的中间阶段获得关于记录数量或其他数据的偶尔引用的状态检查。

Airflow 是否默认在集群中设置 Python 记录器？如果是，我该如何访问它？

Answer 1

如果您使用 Cloud Composer 运行 Airflow，您应该知道 Cloud Composer 仅包含 Airflow 日志和 Streaming 日志。

当您在 Dataproc 集群中运行 PySpark 作业时，作业驱动程序输出存储在 Cloud Storage 中（请参阅Accessing job driver output）

您还可以启用 Dataproc 以将作业驱动程序日志保存在 Cloud Logging 中。

如Dataproc documentation 中所述，要在 Logging 中启用作业驱动程序日志，请在创建集群时设置以下集群属性：

dataproc:dataproc.logging.stackdriver.job.driver.enable=true

以下集群属性也是必需的，并且在创建集群时默认设置：

dataproc:dataproc.logging.stackdriver.enable=true
dataproc:jobs.file-backed-output.enable=true