“容器因超出内存限制而被纱线杀死。在内存为75GB的EMR集群上使用了10.4GB物理内存。”

Question

我正在AWS EMR上运行一个5节点的Spark集群，每个集群的大小为m3.xLarge（1主、4从）。我成功地运行了一个146MB的bzip2压缩CSV文件，最终得到了一个完美的聚合结果。

现在我正试图在这个集群上处理一个~5GB的bzip2 CSV文件，但收到以下错误：

[
 {
  "classification":"spark-env",
  "properties":{

  },
  "configurations":[
     {
        "classification":"export",
        "properties":{
           "PYSPARK_PYTHON":"python34"
        },
        "configurations":[

        ]
     }
  ]
},
{
  "classification":"spark",
  "properties":{
     "maximizeResourceAllocation":"true"
  },
  "configurations":[

  ]
 }
]

def sessionize(raw_data, timeout):
# https://www.dataiku.com/learn/guide/code/reshaping_data/sessionization.html
    window = (pyspark.sql.Window.partitionBy("user_id", "site_id")
              .orderBy("timestamp"))
    diff = (pyspark.sql.functions.lag(raw_data.timestamp, 1)
            .over(window))
    time_diff = (raw_data.withColumn("time_diff", raw_data.timestamp - diff)
                 .withColumn("new_session", pyspark.sql.functions.when(pyspark.sql.functions.col("time_diff") >= timeout.seconds, 1).otherwise(0)))
    window = (pyspark.sql.Window.partitionBy("user_id", "site_id")
              .orderBy("timestamp")
              .rowsBetween(-1, 0))
    sessions = (time_diff.withColumn("session_id", pyspark.sql.functions.concat_ws("_", "user_id", "site_id", pyspark.sql.functions.sum("new_session").over(window))))
    return sessions
def aggregate_sessions(sessions):
    median = pyspark.sql.functions.udf(lambda x: statistics.median(x))
    aggregated = sessions.groupBy(pyspark.sql.functions.col("session_id")).agg(
        pyspark.sql.functions.first("site_id").alias("site_id"),
        pyspark.sql.functions.first("user_id").alias("user_id"),
        pyspark.sql.functions.count("id").alias("hits"),
        pyspark.sql.functions.min("timestamp").alias("start"),
        pyspark.sql.functions.max("timestamp").alias("finish"),
        median(pyspark.sql.functions.collect_list("foo")).alias("foo"),
    )
    return aggregated
 spark_context = pyspark.SparkContext(appName="process-raw-data")
spark_session = pyspark.sql.SparkSession(spark_context)
raw_data = spark_session.read.csv(sys.argv[1],
                                  header=True,
                                  inferSchema=True)
# Windowing doesn't seem to play nicely with TimestampTypes.
#
# Should be able to do this within the ``spark.read.csv`` call, I'd
# think. Need to look into it.
convert_to_unix = pyspark.sql.functions.udf(lambda s: arrow.get(s).timestamp)
raw_data = raw_data.withColumn("timestamp",
                               convert_to_unix(pyspark.sql.functions.col("timestamp")))
sessions = sessionize(raw_data, SESSION_TIMEOUT)
aggregated = aggregate_sessions(sessions)
aggregated.foreach(save_session)

它从这些错误中的一些开始，并停止相同错误的数量增加。

我试着用--conf spark.yarn.executor.memoryeadher运行spark-submit，但这似乎也不能解决问题。

Answer 1

我感觉到你的痛苦..

我们也有类似的问题，即在纱线上使用Spark时内存不足。我们有五个64GB、16个核心VM，无论我们将spark.yarn.executor.memoryoverhead设置为什么，我们都无法为这些任务获得足够的内存--无论我们给它们多少内存，它们最终都会死亡。这是一个相对简单的Spark应用程序，导致了这种情况的发生。

我们发现VM上的物理内存使用率很低，但虚拟内存使用率非常高（尽管日志抱怨物理内存）。我们将yarn.nodemanager.vmem-check-enabled中的yarn-site.xml设置为false，容器不再被杀死，应用程序似乎按预期工作。

该页面有一个来自IBM的非常有用的页面链接:https://web.archive.org/web/20170703001345/https://www.ibm.com/developerworks/community/blogs/kevgrig/entry/linux_glibc_2_10_rhel_6_malloc_may_show_excessive_virtual_memory_usage？lang=en

总之，glibc>2.10改变了它的内存分配。尽管大量的虚拟内存被分配并不是世界末日，但它不能与YARN的默认设置一起工作。

在hadoop-env.sh中将yarn.nodemanager.vmem-check-enabled设置为false，而不是将malloc_arena_max环境变量设置为低值。此bug报告提供了有关此问题的有用信息:https://issues.apache.org/jira/browse/hadoop-7154

我建议通读两页--这些信息非常方便。