4

我正在尝试使用 Apache Spark 来处理我的大型（约 230k 条目）cassandra 数据集，但我经常遇到不同类型的错误。但是，在数据集 ~200 个条目上运行时，我可以成功运行应用程序。我有 3 个节点的 spark 设置，1 个 master 和 2 个 worker，2 个 worker 还安装了 cassandra 集群，数据索引的复制因子为 2。我的 2 个 spark worker 在 Web 界面上显示 2.4 和 2.8 GB 内存，我spark.executor.memory在运行应用程序时设置为 2409，以获得 4.7 GB 的组合内存。这是我的 WebUI 主页

在此处输入图像描述

其中一项任务的环境页面

环境

在这个阶段，我只是尝试使用 spark 处理存储在 cassandra 中的数据。这是我在 Java 中用来执行此操作的基本代码

SparkConf conf = new SparkConf(true)
        .set("spark.cassandra.connection.host", CASSANDRA_HOST)
        .setJars(jars);

SparkContext sc = new SparkContext(HOST, APP_NAME, conf);
SparkContextJavaFunctions context = javaFunctions(sc);

CassandraJavaRDD<CassandraRow> rdd = context.cassandraTable(CASSANDRA_KEYSPACE, CASSANDRA_COLUMN_FAMILY);

System.out.println(rdd.count());

为了成功运行，在一个小数据集（200 个条目）上，事件界面看起来像这样

在此处输入图像描述

但是当我在大型数据集上运行相同的东西（即我只更改CASSANDRA_COLUMN_FAMILY）时，工作永远不会在终端内终止，日志看起来像这样

在此处输入图像描述

大约 2 分钟后，执行者的标准错误看起来像这样

在此处输入图像描述

约7分钟后，我得到

Exception in thread "main" java.lang.OutOfMemoryError: GC overhead limit exceeded

在我的终端中，我必须手动终止该SparkSubmit进程。但是，大型数据集是从仅占用 22 MB 的二进制文件中索引的，并且nodetool status我可以看到只有约 115 MB 的数据存储在我的两个 cassandra 节点中。我也尝试在我的数据集上使用 Spark SQL，但也得到了类似的结果。对于 Transformation-Action 程序和使用 Spark SQL 的程序，我的设置哪里出错了，我应该怎么做才能成功处理我的数据集。

我已经尝试过以下方法

  • 用于-Xms1G -Xmx1G增加内存，但程序失败并出现异常说我应该改为 set spark.executor.memory，我有。

  • 使用spark.cassandra.input.split.size，它不能说它不是一个有效的选项，类似的选项是spark.cassandra.input.split.size_in_mb，我设置为 1，没有效果。

编辑

基于这个答案，我还尝试了以下方法：

  • 设置spark.storage.memoryFraction为 0

  • 未设置spark.storage.memoryFraction为零并与persist、MEMORY_ONLY和MEMORY_ONLY_SER一起使用。MEMORY_AND_DISKMEMORY_AND_DISK_SER

版本：

  • 火花：1.4.0

  • 卡桑德拉：2.1.6

  • 火花卡桑德拉连接器：1.4.0-M1

4

1 回答 1

6

我认为最新的 spark-cassandra-connector 存在问题。该参数spark.cassandra.input.split.size_in_mb的默认值为 64 MB，在代码中被解释为 64 字节。这会导致创建的分区太多，spark无法调度。尝试将 conf 值设置为

spark.cassandra.input.split.size_in_mb=67108864
于 2015-06-25T11:13:12.790 回答