AWS托管Apache Airflow DAG日志查询求助：EMR任务日志获取繁琐

1. 扩展Airflow Operator自动拉取日志到任务日志 #

扩展EmrAddStepsOperator或EmrStepSensor，添加失败回调逻辑，直接从S3拉取关键日志内容并输出到Airflow任务日志中，无需手动跳转S3。

示例代码片段：

from airflow.providers.amazon.aws.operators.emr import EmrAddStepsOperator
import boto3

def pull_emr_step_logs(context):
    s3_client = boto3.client('s3')
    cluster_id = context['task_instance'].xcom_pull(key='cluster_id')
    step_id = context['task_instance'].xcom_pull(key='step_id')
    log_bucket = 'test-bucket-logs'
    log_prefix = f"{cluster_id}/steps/{step_id}/"
    
    # 列出S3日志路径下的文件
    response = s3_client.list_objects_v2(Bucket=log_bucket, Prefix=log_prefix)
    for obj in response.get('Contents', []):
        # 优先拉取stderr或含错误关键字的日志文件
        if 'stderr' in obj['Key'] or 'error' in obj['Key'].lower():
            log_obj = s3_client.get_object(Bucket=log_bucket, Key=obj['Key'])
            print(f"=== 日志文件: {obj['Key']} ===")
            print(log_obj['Body'].read().decode('utf-8'))

# 定义Operator时绑定失败回调
emr_step_operator = EmrAddStepsOperator(
    task_id='test',
    job_flow_id='{{ task_instance.xcom_pull(key="cluster_id") }}',
    steps=[...],
    on_failure_callback=pull_emr_step_logs,
    dag=dag
)

2. 配置EMR集群同步日志到CloudWatch #

创建EMR集群时开启CloudWatch日志集成，将步骤日志自动同步到CloudWatch Logs：

• 在EMR集群配置的LogUri字段指定S3日志路径的同时，启用CloudWatchLogConfiguration，设置自定义日志组（比如/aws/emr/clusters）。
• 任务失败后，直接在CloudWatch Logs中通过集群ID或步骤ID搜索日志，无需逐层遍历S3文件夹。

3. 使用AWS CLI快速查询EMR步骤日志 #

用aws emr logs命令直接拉取指定步骤的日志内容，跳过手动操作S3的步骤：

# 拉取指定步骤的全部日志
aws emr logs get --cluster-id j-HNKNG13GHYTD --step-id s-UYHJGTHEFGER

# 仅拉取stderr日志
aws emr logs get --cluster-id j-HNKNG13GHYTD --step-id s-UYHJGTHEFGER --stderr

4. 预定义S3日志路径的快速访问逻辑 #

在Airflow DAG中，将EMR步骤的完整日志文件路径（比如容器日志的具体路径）通过XCom存储，任务失败时直接在Airflow UI的XCom面板查看，或结合boto3自动打印关键日志路径。

内容的提问来源于stack exchange，提问作者dhanabalan Rangasamy