定期删除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 steps日志并存储在路径/mnt/var/logs/hadoop/steps/step_id/上，在创建集群时，我给出了一个logUri路径，日志根据该路径复制到s3位置。所以我的猜测是，由于这些日志，步骤失败正在发生。

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

Answer 1

您可以使用以下boto3代码（我确信这也可以在Java中用AWS SDK for Java完成）来删除日志，对于定期删除，您有如下选项

1. 使用像Airflow这样的工作流调度程序，请参见下面的示例
2. 使用它作为lambda函数并安排它定期运行（容易得多）
3. 本地使用cron jon（不太可行）

函数删除日志（输入即将到期的阈值、桶名和前缀，可以是“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

  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调度它，请参见此处的lambda调度指南