java - Apache Spark 无法处理大型 Cassandra 列族

Question

我正在尝试使用 Apache Spark 来处理我的大型(约 230k 个条目)cassandra 数据集，但我不断遇到各种错误。但是，在约 200 个条目的数据集上运行时，我可以成功运行应用程序。我有一个包含 3 个节点的 spark 设置，其中有 1 个主节点和 2 个工作节点，这 2 个工作节点还安装了一个 cassandra 集群，其数据索引的复制因子为 2。我的 2 个 spark 工作节点在 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 分钟后，执行程序的 stderr 看起来像这样

大约 7 分钟后，我得到了

线程“main”中的异常 java.lang.OutOfMemoryError:超出 GC 开销限制


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

我已经尝试过以下方法

• 使用 -Xms1G -Xmx1G 增加内存，但程序失败并出现异常提示我应该改为设置 spark.executor.memory，我有。

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

编辑

基于 this回答，我也试过以下方法:

• 将 spark.storage.memoryFraction 设置为 0

• 不将 spark.storage.memoryFraction 设置为零，并将 persist 与 MEMORY_ONLY、MEMORY_ONLY_SER 一起使用、MEMORY_AND_DISK 和 MEMORY_AND_DISK_SER。

版本:

• Spark:1.4.0

• Cassandra :2.1.6

• spark-cassandra-连接器:1.4.0-M1

Answer 1

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

spark.cassandra.input.split.size_in_mb=67108864