aws emr spark app tuning for groupby with>300000个组

步骤:

1)从s3读取文件(140-200gb)。

2)然后我应用filter()函数,删除大多数记录的映射(大约70%将被删除)。

3)对于剩余的记录(总共大约40-50GB),使用map()创建myobj的javardd。

4)之后,我使用group by按sessisonid字段对对象进行分组。对于40-50GB的总容量,我预计groupby中会有大约300000个组。

5)然后,对于每个组,我执行processSingGroupOfEvents()(此函数如下:它接受Interable,然后为每个Iterable(总共300000个)执行一些简单的staff。

6)之后,我使用coalesce()和saveastextfile将结果输出到s3(输出文件大小约为1-2gb)。

Pseudocode:

 JavaRDD<MyObj> eventsCollection = context.textFile(input)
                .filter(data -> applySomeFilter(data))      // This will filter ~70% of records 
                .map(data ->  createMyObjFromData(data));

JavaPairRDD<String, Iterable<MyObj>> eventsCollectionGroupedById = eventsCollection
                .groupBy(x -> x.getSessionId())
                .persist(StorageLevel.MEMORY_AND_DISK());;

JavaPairRDD<String, String> groupedByIdResults = eventsCollectionGroupedById
                .mapValues(iterable -> processSingeGroupOfEvents(iterable, additionalVariable1, additionalVariable2 ));

groupedByIdResults
                .coalesce(1) // Union all partitions into single one (in order to avoid multiple output files)
                .map(data ->  data._2()) 
                .saveAsTextFile(outputS3Location);

我目前对AWS EMR Spark应用的配置是:

• R3.8XL核心实例类型中的4个
• 主节点为R3.8XL中的1个
• EMR发布标签:EMR-5.11.0
• MaximizeResourceAllocation=true。
• Spark版本1.4(我现在无法将Spark更新到最新版本)

目前,执行这样的工作大约需要30-50分钟。但是,在将来,我预计输入数据的大小将增加一倍(大约300 GB的数据=>~600000个组在会话ID上聚合)

我如何估计,如果我的数据加倍,我的集群将能够处理这样的负载?此外,有时我会错误地说R3.8XL实例类型超出了我所在区域的配额限制,所以我担心如果添加更多硬件,我将更经常遇到此问题。

更新: processSingGroupOfevents()方法在具有相同会话ID的事件组(iterable)上迭代,并执行一些复杂的计算(例如,计算运行总数、查找组中某些元素的最大值、分析时间戳等)。它返回逗号分隔的字符串,其中包含特定会话ID的聚合值。