小编Fur*_*rge的帖子

让我先说我是斯卡拉的新手; 但是,我发现基于Actor的并发模型很有趣,我试着给它一个相对简单的应用程序.我遇到的问题是,虽然我能够让应用程序工作,但结果远远低于基于Java的等效解决方案的效率(就实时,CPU时间和内存使用而言).使用从ArrayBlockingQueue中提取消息的线程.我想明白为什么.我怀疑这可能是我缺乏Scala知识,而且我造成了所有低效率,但经过多次尝试重新编写应用程序但没有成功,我决定联系社区寻求帮助.

我的问题是:我有一个gzip压缩文件,其格式为:

SomeID comma_separated_list_of_values

例如:

1234 12,45,82

我想解析每一行,并得到逗号分隔列表中每个值的出现次数的总体计数.

此文件可能非常大(几GB压缩),但每个文件的唯一值数量非常小(最多500个).我认为这是一个尝试编写基于Actor的并发Scala应用程序的好机会.我的解决方案涉及一个主驱动程序,它创建一个解析器Actors池.然后主驱动程序从stdin读取行,将行传递给一个解析行的Actor并保持值的本地计数.当主驱动程序读取最后一行时,它会向每个actor传递一条消息,指示已读取所有行.当演员收到"完成"消息时,他们将计数传递给聚合器,该聚合器对所有演员的计数进行求和.一旦聚合了所有解析器的计数,主驱动程序就会打印出统计信息.

问题:我遇到的主要问题是这个应用程序的低效率.与使用线程和ArrayBlockingQueue的"等效"Java应用程序相比,它使用了更多的CPU和更多的内存.为了正确看待这一点,我收集了一些1000万行测试输入文件的统计数据:

Scala 1 Actor(解析器):

    real    9m22.297s
    user    235m31.070s
    sys     21m51.420s

Java 1线程(解析器):

    real    1m48.275s
    user    1m58.630s
    sys     0m33.540s

Scala 5演员:

    real    2m25.267s
    user    63m0.730s
    sys     3m17.950s

Java 5主题:

    real    0m24.961s
    user    1m52.650s
    sys     0m20.920s

此外,top报告称Scala应用程序的驻留内存大小约为10倍.所以我们谈论的是更多的CPU和内存量级,以达到数量级更差的性能,而我无法弄清楚导致这种情况的原因.这是一个GC问题,还是我以某种方式创建了比我意识到的更多的对象副本？

其他可能重要或可能不重要的细节:

• scala应用程序由Java类包装,因此我可以提供一个自包含的可执行JAR文件(我在每台可能要运行此应用程序的计算机上都没有Scala jar).
• 正在调用该应用程序如下:gunzip -c gzFilename | java -jar StatParser.jar

这是代码:

主要司机:

import scala.actors.Actor._
import scala.collection.{ immutable, mutable }
import scala.io.Source

class StatCollector (numParsers : Int ) {
    private val parsers = new mutable.ArrayBuffer[StatParser]()
    private …

我是猪的新手,我在解析输入并将其转换为我可以使用的格式时遇到问题.输入文件包含具有固定字段和KV对的行,如下所示:

FF1 | FF2 | FF3 | FF4 | KVP1 | KVP2 | ... | KVPn

我的目标是计算每个KV对的唯一固定字段组合的数量.所以考虑以下输入行:

1|2|3|4|key1=value1|key2=value2
2|3|4|5|key1=value7|key2=value2|key3=value3

当我完成后,我希望能够生成以下结果(此时输出格式并不重要,我只是向您展示我想要的结果):

key1=value1 : 1
key1=value7 : 1
key2=value2 : 2
key3=value3 : 1

似乎我应该能够通过将固定区域分组并展平一袋KV对来生成交叉产品

我试过用以下的东西读这个:

data = load 'myfile' using PigStorage('|');
A = foreach data generate $0 as ff1:chararray, $1 as ff2:long, $2 as ff3:chararray, $3 as ff4:chararray, TOBAG($4..) as kvpairs:bag{kvpair:tuple()};
B = foreach A { sorted = order A by ff2; lim = limit sorted 1; generate group.ff1, group.ff4, flatten( …