定期删除 EMR 集群日志答案

【问题标题】：Removing EMR cluster logs periodically定期删除 EMR 集群日志
【发布时间】：2020-09-04 19:39:40
【问题描述】：

我有一个 EMR 集群，它成功运行了一个 spark 流作业几天。但几天后，集群因步骤失败而终止。我检查了日志，它说

OpenJDK 64-Bit Server VM warning: INFO: os::commit_memory(0x00007f8cb0854000, 12288, 0) failed; error='Cannot allocate memory' (errno=12)
Command exiting with ret '1'

对于这个错误，我查了一下发现，对于JRE来说内存不够。
我发现集群会创建 EMR 步骤日志并存储在路径 /mnt/var/logs/hadoop/steps/step_id/ 上，并且在创建集群时我已经给出了一个 logUri 路径因此，日志被复制到 s3 位置。所以我的猜测是，由于这些日志，步骤失败正在发生。

谁能建议我如何定期从集群中删除这些 emr 步骤日志，以免集群内存不足？

【问题讨论】：

标签： apache-spark pyspark amazon-emr

【解决方案1】：

您可以使用以下 boto3 代码（我相信这也可以使用适用于 Java 的 AWS 开发工具包在 Java 中完成）来删除日志，为了定期删除，您可以选择

使用 Airflow 等工作流调度程序，请参见下面的示例
将其用作 lambda 函数并安排它定期运行（更容易）
在本地使用 cron jon（不太可行）

日志删除功能（输入即将过期的threshold、bucket name和prefix，可以是"logs/sparksteps/j-"）

def clean_s3(buck, match_prefix,exp_threshold):

    s3_client = boto3.client('s3')
    key_names = []
    file_timestamp = []
    file_size = []
    kwargs = {"Bucket": buck, "Prefix": match_prefix}
    while True:
        result = s3_client.list_objects_v2(**kwargs)
        for obj in result["Contents"]:

            if "." in obj["Key"]:
                key_names.append(obj["Key"])
                file_timestamp.append(obj["LastModified"].timestamp())
                file_size.append(obj["Size"])
        try:
            kwargs["ContinuationToken"] = result["NextContinuationToken"]
        except KeyError:
            break

    key_info = {
        "key_path": key_names,
        "timestamp": file_timestamp,
        "size": file_size
    }
    #print(f'All Keys in {buck} with {prefix} Prefix found!')
    s3_file = key_info
    for i, fs in enumerate(s3_file["timestamp"]):
            #file_expired = is_expired(fs)
            #print(fs)
            if fs < exp_threshold: #if True is recieved
                    print("Deleting %s" % {s3_file["key_path"][i]})
                    s3_client.delete_object(Bucket=buck, Key=s3_file["key_path"][i])

您可以计算需要通过的到期阈值（以纪元秒为单位），如下所示

date_now = time.time()
days = 7 # 7 days
total_time = 86400*days 
exp_threshold = date_now-total_time

现在，对于选项 1，您可以制作如下所示的气流运算符

  s3_cleanup = PythonOperator(
        task_id='s3cleanup',
        python_callable=clean_s3,
        op_kwargs={
            'buck': '<you bucket>',
            'match_prefix': "logs/sparksteps/j-",
            'exp_threshold':exp_threshold,
            
    },dag=dag)

或者，使用 apporach 2，您可以使用 AWS lamda 进行调度，请参阅 guide for schedling with lambda here

【讨论】：