xml - 使用拉式解析器的 Scala 内存泄漏

Question

我一直在尝试编写一个 XML 解析器来读取维基百科 XML 转储(英语，只有当前修订版，压缩后大约 6.2Gb)并且一直在使用 Scala 2.8.1 pull 解析器。它得到了一个合理的通过(超过 1000 万篇文章中的 300 万篇)但似乎逐渐泄漏内存并最终因堆外错误而爆炸。我将堆增加到 1.5Gb 并且它变得更远(几乎到最后)，但后来我得到(我忘记了确切的异常)一个错误表明垃圾收集器正在放弃(花费了整个处理资源的很大一部分无需回收太多)。

我的代码对我来说似乎是合理的(尽管它还不是惯用的函数式 scala)而且我看不到任何明显的泄漏源。我也知道拉式解析器仍在完善中——但我太清楚自己的无知，无法将其称为库问题。我是一位经验丰富的 C++ 和 Python 程序员，但我刚刚接触 Scala，因此非常感谢任何反馈。

import java.io.{FileInputStream, BufferedInputStream}
import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream
import org.apache.hadoop.io.SequenceFile.{createWriter}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.io.Text
import org.apache.hadoop.io.SequenceFile.CompressionType.BLOCK

import scala.io.Source
import scala.xml.pull.{XMLEventReader, EvElemStart, EvElemEnd, EvText}



object Crunch
{
    private def parsePage( parser : XMLEventReader ) : (String, Long, Long, String) =
    {
        var title = ""
        var id = 0
        var revision = 0
        var text = ""
        var done = false
        while ( parser.hasNext && !done )
        {
            parser.next match
            {
                case EvElemStart(_, "title", _, _ ) =>
                {
                    title = getText( parser, "title" )
                }
                /*case EvElemStart(_, "revision", _, _) =>
                {
                    // Need to get the 'id' from revision
                    revision = getText( parser, "revision" ).toInt
                }*/
                case EvElemStart(_, "id", _, _ ) =>
                {
                    id = getText( parser, "id" ).toInt
                }
                case EvElemStart(_, "text", _, _ ) =>
                {
                    text = getText( parser, "text" )
                }
                case EvElemEnd(_, "page") =>
                {
                    done = true
                }
                case _ =>
            }
        }
        return (title, id, revision, text)
    }

    private def getText( parser : XMLEventReader, inTag : String ) : String =
    {
        var fullText = new StringBuffer()
        var done = false
        while ( parser.hasNext && !done )
        {
            parser.next match
            {
                case EvElemEnd(_, tagName ) =>
                {
                    assert( tagName.equalsIgnoreCase(inTag) )
                    done = true
                }
                case EvText( text ) =>
                {
                    fullText.append( text )
                }
                case _ =>
            }
        }
        return fullText.toString()
    }
    def main( args : Array[String] )
    {
        require( args.length == 2 )
        val fin = new FileInputStream( args(0) )
        val in = new BufferedInputStream(fin)
        val decompressor = new BZip2CompressorInputStream(in)

        val runtime = Runtime.getRuntime

        val conf = new Configuration()
        val fs = FileSystem.get(conf)        

        //val writer = createWriter( fs, conf, new Path(args(1)), new Text().getClass(), new Text().getClass(), BLOCK )

        var count = 0
        try
        {
            val source = Source.fromInputStream( decompressor )
            val parser = new XMLEventReader(source)

            while (parser.hasNext)
            {
                parser.next match
                {
                    case EvElemStart(_, "page", attrs, _) =>
                    {
                        val (title, id, revision, text) = parsePage( parser )

                        //writer.append( new Text(title), new Text(text) )

                        count = count + 1
                        if ( count % 100 == 0 )
                        {
                            printf("%s %d (%dMb mem, %dMb free)\n", title, count,
                                (runtime.totalMemory/1024/1024).toInt,
                                (runtime.freeMemory/1024/1024).toInt )
                        }
                    }
                    case _ =>
                }
                // Do something
            }
        }
        finally
        {
            decompressor.close()
            fin.close()
        }

        println( "Finished decompression.")
    }
}

Answer 1

在 trunk 中修复了 XML pull 解析器的 2 种类型的内存问题:

1. CData和 processing instruction阻止垃圾收集的元素
2. Elements with a lot of children ，每个 child 占用一点内存，最终堆耗尽。

第一个问题通常会导致非常快的内存不足问题，所以不太可能。

两者都应该每晚修复 2.9.0，我建议使用它。如果你运行在 2.9.0 issues 因为它是 trunk 并且可能不稳定，你也可以通过下载并在本地编译 XMLEventEventReader 和 MarkupParser 来反向移植这两个补丁，然后将输出打包为 00patch.jar，使其位于 scala libs jar 之前，并将其放在 2.8.1 安装的 $SCALA_HOME/lib 下。