Spark LDA消耗太多内存

Question

我正在尝试使用spark mllib lda来总结我的文档语料库.

我的问题设置如下.

• 大约100,000份文件
• 大约400,000个独特的单词
• 100集群

我有16台服务器(每台有20个内核和128GB内存).

当我执行LDA OnlineLDAOptimizer,它给了内存不足的错误,提示我增加spark.driver.maxResultSize类似 的11个任务(1302 MB)序列化结果总大小比spark.driver.maxResultSize大

我增加到spark.driver.maxResultSize120GB(也spark.driver.memory增加到120GB)并重新运行LDA但不缺.它仍然说11个任务(120.1 GB)的序列化结果的总大小比spark.driver.maxResultSize大

我尝试了另外一个包含大约100,000个独特单词的数据集,并且它有效

那么,在使用Spark mllib LDA时,如何估计内存使用量？我在官方文档中找不到任何规范.

注意我使用稀疏向量来构造RDD[(Long, Vector)]传递给docuemnt ,LDA.run()但不知道spark lda是否可以在内部正确处理稀疏格式.

(编辑)我使用Scala版本的LDA.不是Python版本.

这可能是一个相关的问题,但没有给出明确的答案. Spark LDA困境 - 预测和OOM问题

(编辑)的

这是我的代码片段(要点). https://gist.github.com/lucidfrontier45/11420721c0078c5b7415

def startJob(args: RunArgs)(implicit sc: SparkContext): Unit = {
    val src = sc.textFile(args.fname, minPartitions = args.n_partitions).map(_.split("\t"))
        .flatMap {
            // input file's format is (user_id, product_name, count)
            case Array(u, p, r, t) => Some((u.toInt, p.toInt, r.toDouble))
            case _ => None
        }.persist()

    // Map to convert user_id or product_name into unique sequencential id
    val userid_map = src.map(_._1).distinct().zipWithIndex().collect().toMap
    val productid_map = src.map(_._2).distinct().zipWithIndex().collect().toMap
    val inverse_userid_map = userid_map.map(_.swap)

    // broadcat to speedup RDD map operation
    val b_userid_map = sc.broadcast(userid_map)
    val b_productid_map = sc.broadcast(productid_map)
    val b_inverse_userid_map = sc.broadcast(inverse_userid_map)

    // run map
    val transformed_src = src.map { case (u, p, r) =>
        (b_userid_map.value(u), b_productid_map.value(p).toInt, r)
    }

    println("unique items = %d".format(b_productid_map.value.size))

    // prepare for LDA input RDD[(LONG, Vector)]
    val documents = transformed_src.map { case (u, p, r) => (u, (p, r)) }
        .groupByKey()
        .map { t => (t._1, Vectors.sparse(b_productid_map.value.size, t._2.toSeq)) }.persist()

    documents.count()
    src.unpersist()

    // run Online Variational LDA
    val ldamodel = new LDA()
        .setK(args.k)
        .setMaxIterations(args.n_iter)
        .setOptimizer("online")
        .run(documents)
        .asInstanceOf[LocalLDAModel]


    val result = ldamodel.topicDistributions(documents)
        .map { case (i, v) =>
            val u = b_inverse_userid_map.value(i)
            "%d,%s".format(u, v.toArray.mkString(","))
        }
    result.saveAsTextFile(args.out)
}

实际上,我使用LDA来减少交易数据的尺寸.我的数据是格式(u, p, r) ,其中u是用户ID,p是产品名称,r是用户数u与互动p.在这种情况下,user对应于文档和产品.由于用户标识和产品名称是任意字符串,因此在提交给LDA之前,我将它们转换为唯一的顺序整数.

谢谢.

Answer 1

造成此问题的常见原因有三个，它们可能独立发生，也可能同时发生。

1. 该作业使用诸如collect. 唉，一些 SparkML 代码就是这样做的。如果您不能将问题归咎于下面的 (2) 或 (3)，那么问题很可能是您的数据与实现交互的结果OnlineLDAOptimizer。

2. 该作业涉及大量任务，每个任务都会将结果返回给驱动程序，作为 Spark 作业管理的一部分（而不是类似的任务collect）。检查SparkUI中的任务数。另请参阅超出“spark.driver.maxResultSize”而不将任何数据带入驱动程序是否org.apache.spark.scheduler.TaskSetManager#canFetchMoreResults在org.apache.spark.scheduler.TaskResultGetter#enqueueSuccessfulTask堆栈跟踪中？

3. 估计错误：Spark 显着高估了将要返回给驱动程序的数据大小，并抛出此错误，以防止集群驱动程序出现 OOM。请参阅什么是spark.driver.maxResultSize？测试此问题的一种方法是将其设置spark.driver.maxResultSize为 0（无限制），然后看看会发生什么。

希望这可以帮助！