Apache Spark 无法处理大型 Cassandra 列族

Question

我正在尝试使用 Apache Spark 来处理我的大型（约 230k 个条目）cassandra 数据集，但我不断遇到不同类型的错误。但是，在大约 200 个条目的数据集上运行时，我可以成功运行应用程序。我有 3 个节点的 spark 设置，其中包含 1 个主节点和 2 个工作线程，并且 2 个工作线程还安装了一个 cassandra 集群，其中数据索引的复制因子为 2。我的 2 个 spark 工作线程在 Web 界面上显示 2.4 GB 和 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 分钟后，我得到

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增加内存，但该计划失败，一个异常说我应该改为设置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_DISK和MEMORY_AND_DISK_SER。

版本：

• 火花：1.4.0

• 卡桑德拉：2.1.6

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

Answer 1

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

spark.cassandra.input.split.size_in_mb=67108864