定期删除EMR群集日志

Question

我有一个EMR集群，它成功地运行了几天的火花流作业。但几天后，集群因步骤故障而终止。我查过日志，上面写着

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日志并存储在path /mnt/var/logs/hadoop/steps/step_id/上，在创建集群时，我给出了一个logUri路径，由于该路径将日志复制到s3位置。因此，我猜想，由于这些日志，步进故障正在发生。

有人能建议我如何定期从集群中删除这些emr步骤日志，这样集群就不会耗尽内存了吗？

Answer 1

您可以使用下面的boto3代码(我确信这可以用AWS在Java中完成)来删除日志，对于定期删除，您可以选择如下

1. 使用工作流调度程序(如气流)，请参见下面的示例
2. 使用它作为lambda函数，并安排它定期运行(容易得多)
3. 在本地使用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来调度它，请参阅这里使用lambda的调度指南