airflow.providers.amazon.aws.log.cloudwatch_task_handler¶

类¶

`CloudWatchRemoteLogIO`	配置有类名的日志记录器的便捷超类。
`CloudwatchTaskHandler`	CloudwatchTaskHandler 是一个 Python 日志处理器，用于处理和读取任务实例日志。

函数¶

`json_serialize_legacy`(value)	复制旧版 watchtower 行为的 JSON 序列化器。
`json_serialize`(value)	复制当前 watchtower 行为的 JSON 序列化器。

模块内容¶

airflow.providers.amazon.aws.log.cloudwatch_task_handler.json_serialize_legacy(value)[source]¶

复制旧版 watchtower 行为的 JSON 序列化器。

继承自 watchtower@2.0.1 的 JSON 序列化函数，将 datetime 对象序列化为 ISO 格式，将所有其他非 JSON 可序列化对象序列化为 null。

参数:: value (Any) – 要序列化的对象
返回值:: 如果 value 是 datetime 实例，则返回其字符串表示形式，否则返回 None
返回类型:: str | None

airflow.providers.amazon.aws.log.cloudwatch_task_handler.json_serialize(value)[source]¶

复制当前 watchtower 行为的 JSON 序列化器。

这为用户提供了一个可访问的导入路径：airflow.providers.amazon.aws.log.cloudwatch_task_handler.json_serialize

参数:: value (Any) – 要序列化的对象
返回值:: value 的字符串表示形式
返回类型:: str | None

class airflow.providers.amazon.aws.log.cloudwatch_task_handler.CloudWatchRemoteLogIO(context=None)[source]¶

基类: airflow.utils.log.logging_mixin.LoggingMixin

配置有类名的日志记录器的便捷超类。

base_log_folder: pathlib.Path[source]¶

remote_base: str = ''[source]¶

delete_local_copy: bool = True[source]¶

log_group_arn: str[source]¶

log_stream_name: str = ''[source]¶

log_group: str[source]¶

region_name: str[source]¶

property hook[source]¶: 返回 AwsLogsHook。

property handler: watchtower.CloudWatchLogHandler[source]¶

property processors: tuple[structlog.typing.Processor, Ellipsis][source]¶

close()[source]¶

upload(path, ti)[source]¶

read(relative_path, ti)[source]¶

get_cloudwatch_logs(stream_name, task_instance)[source]¶

返回给定日志流中的所有日志。

参数:

stream_name (str) – 要获取所有日志的 Cloudwatch 日志流名称
task_instance (airflow.sdk.types.RuntimeTaskInstanceProtocol) – 要获取日志的任务实例

返回值:

给定日志流中所有日志的字符串形式

class airflow.providers.amazon.aws.log.cloudwatch_task_handler.CloudwatchTaskHandler(base_log_folder, log_group_arn, **kwargs)[source]¶

基类: airflow.utils.log.file_task_handler.FileTaskHandler, airflow.utils.log.logging_mixin.LoggingMixin

CloudwatchTaskHandler 是一个 Python 日志处理器，用于处理和读取任务实例日志。

它扩展了 Airflow FileTaskHandler，并上传日志到 Cloudwatch 并从 Cloudwatch 读取日志。

参数:

base_log_folder (str) – 本地存储日志的基础文件夹
log_group_arn (str) – 用于远程日志存储的 Cloudwatch 日志组的 ARN，格式为 arn:aws:logs:{region name}:{account id}:log-group:{group name}

trigger_should_wrap = True[source]¶

handler = None[source]¶

log_group[source]¶

region_name[source]¶

closed = False[source]¶

io[source]¶

property hook[source]¶: 返回 AwsLogsHook。

set_context(ti, *, identifier=None)[source]¶

为 Airflow 任务处理器提供任务实例上下文。

通常返回 None。但如果属性 maintain_propagate 已设置为 propagate，则返回 sentinel MAINTAIN_PROPAGATE。这会覆盖默认行为，即在每次调用 set_context 时将 propagate 设置为 False。在撰写本文时，此功能仅用于单元测试。

参数:

ti (airflow.models.taskinstance.TaskInstance) – 任务实例对象
identifier (str | None) – 如果设置，则向日志文件名添加后缀。当从任务或触发器运行以外的上下文将异常消息中继到任务日志时使用。

close()[source]¶: 关闭负责将本地日志文件上传到 Cloudwatch 的处理器。