Apache Spark 无法处理大型 Cassandra 列族

Question

我正在尝试使用 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 时，我可以看到我的两个 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 一起使用。

版本：

• 火花：1.4.0

• 卡桑德拉：2.1.6

• spark-cassandra-connector：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