使用Spark处理大数据时出错

Question

我正在尝试在AWS上使用Spark进行字数统计。我创建了一个spark集群，并将大文件存储在s3中。我总共有100个文件，每个文件都是50 of。我可以一次成功地对这些文件中的一个文件进行字数统计，但如果我尝试将它们全部计算在一起，则会失败。此外，如果我尝试对大于60 if的单个文件进行字数统计，则会失败。我得到的错误是：

ERROR org.apache.spark.executor.Executor (Executor task launch worker for task 2): Exception in task 2.0 in stage 0.0 (TID 2)
java.lang.NoSuchMethodError: org.roaringbitmap.RoaringBitmap.runOptimize()Z
    at org.apache.spark.scheduler.HighlyCompressedMapStatus$.apply(MapStatus.scala:234)
    at org.apache.spark.scheduler.MapStatus$.apply(MapStatus.scala:54)
    at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:74)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
    at org.apache.spark.scheduler.Task.run(Task.scala:108)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:338)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

这是我的字数统计代码

public class WordCount {

    public static void main(String[] args) throws Exception {
        SparkConf sparkConf = new SparkConf().setAppName("WordCount").setMaster("local[5]").set("spark.executor.memory","32g");
        JavaSparkContext spark = new JavaSparkContext(sparkConf);
        System.out.println(MessageFormat.format("Starting word count for {0}", args[0]));
        JavaRDD<String> textFile = spark.textFile(MessageFormat.format("s3a://my_bucket/{0}*", args[0]));
        JavaPairRDD<String, Integer> counts = textFile.flatMap(s -> Arrays.asList(s.split(" ")).iterator())
                .mapToPair(word -> new Tuple2<>(word, 1)).reduceByKey((a, b) -> a + b);
        counts.saveAsTextFile(MessageFormat.format("{0}_output", args[0]));

        spark.close();
        System.exit(0);
    }
}

我读到这可能是一个版本不匹配错误，我已经保证我编译代码中的spark版本和我集群上的spark版本是相同的。

我的集群有1个主节点和5个核心节点。每个节点具有8核、64 of内存和1690 of存储。

为了运行代码，我将上面的java代码导出到一个jar文件中。我将jar文件保存到s3。然后在亚马逊网络服务控制台中，我点击"add step“并选择我上传到s3的jar文件。然后运行代码。

Answer 1

@ErnestKiwele评论对我帮助最大。在AWS中创建集群时，请选择高级选项。默认选择Hadoop、Pig、Hive和Hue。我保留了所有选中并添加到Spark中的内容。当我只使用Hadoop和Spark (这是我需要的全部)时，我不再收到错误。一定是spark和其他人之间存在一些冲突。谢谢!