2

我有一个简单的 Spark 作业，从 5 节点 Cassandra 集群读取 500m 行，该集群始终运行 6 个任务，由于每个任务的大小，这会导致写入问题。我试过调整input_split_size，好像没有效果。目前我被迫重新分区表扫描，这并不理想，因为它很昂贵。

阅读了几篇文章后，我尝试在启动脚本（如下）中增加 num-executors，尽管这没有效果。

如果没有办法在 Cassandra 表扫描中设置任务数量，那很好，我会做的。但我有一种持续的琐碎感觉，我在这里遗漏了一些东西。

Spark 工作人员生活在 C* 节点上，这些节点是 8 核、64GB 服务器，每个服务器配备 2TB SSD。

...
val conf = new SparkConf(true).set("spark.cassandra.connection.host",
cassandraHost).setAppName("rowMigration")
  conf.set("spark.shuffle.memoryFraction", "0.4")
  conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  conf.set("spark.executor.memory", "15G")
  conf.set("spark.cassandra.input.split.size_in_mb", "32") //default 64mb
  conf.set("spark.cassandra.output.batch.size.bytes", "1000") //default
  conf.set("spark.cassandra.output.concurrent.writes", "5") //default

val sc = new SparkContext(conf)

val rawEvents = sc.cassandraTable(cassandraKeyspace, eventTable)
  .select("accountid", "userid", "eventname", "eventid", "eventproperties")
  .filter(row=>row.getString("accountid").equals("someAccount"))
  .repartition(100)

val object = rawEvents
  .map(ele => (ele.getString("userid"),
    UUID.randomUUID(),
    UUID.randomUUID(),
    ele.getUUID("eventid"),
    ele.getString("eventname"),
    "event type",
    UUIDs.unixTimestamp(ele.getUUID("eventid")),
    ele.getMap[String, String]("eventproperties"),
    Map[String, String](),
    Map[String, String](),
    Map[String, String]()))
  .map(row=>MyObject(row))

Object.saveToCassandra(targetCassandraKeyspace,eventTable)

启动脚本：

#!/bin/bash
export SHADED_JAR="Migrate.jar"
export SPARKHOME="${SPARKHOME:-/opt/spark}"
export SPARK_CLASSPATH="$SHADED_JAR:$SPARK_CLASSPATH"
export CLASS=com.migration.migrate
"${SPARKHOME}/bin/spark-submit" \
        --class "${CLASS}" \
        --jars $SHADED_JAR,$SHADED_JAR \
        --master spark://cas-1-5:7077  \
        --num-executors 15 \
        --executor-memory 20g \
        --executor-cores 4 "$SHADED_JAR" \
        --worker-cores 20 \
        -Dcassandra.connection.host=10.1.20.201 \
        -Dzookeeper.host=10.1.20.211:2181 \

编辑 - 遵循 Piotr 的回答：

我在 sc.cassandraTable 上设置了 ReadConf.splitCount，如下所示，但这不会改变生成的任务数，这意味着我仍然需要重新分区表扫描。我开始认为我正在考虑这个错误并且重新分区是必要的。目前这项工作大约需要 1.5 小时，将表扫描重新划分为 1000 个任务，每个任务大约 10MB，从而将写入时间减少到几分钟。

val cassReadConfig = new ReadConf {
      ReadConf.apply(splitCount = Option(1000)
        )
    }

    val sc = new SparkContext(conf)

    val rawEvents = sc.cassandraTable(cassandraKeyspace, eventTable)
    .withReadConf(readConf = cassReadConfig)
4

2 回答 2

3

从 spark 连接器 1.3 开始，拆分大小是根据自 Cassandra 2.1.5 以来可用的 system.size_estimates Cassandra 表估计的。该表由 Cassandra 定期刷新，在加载/删除新数据或加入新节点后不久，其内容可能不正确。检查那里的估计是否反映了您的数据量。这是一个相对较新的功能，因此也很可能存在一些错误。

如果估计错误，或者您运行的是较旧的 Cassandra，我们保留了覆盖自动拆分大小调整的功能。sc.cassandraTable 采用 ReadConf 参数，您可以在其中设置 splitCount，这将强制进行固定数量的拆分。

至于split_size_in_mb参数，确实在项目源码中存在一段时间的bug，但是在发布到任何发布到maven的版本之前已经修复了。因此，除非您从（旧）源代码编译连接器，否则不应点击它。

于 2015-07-29T06:03:20.867 回答
0

split.size_in_mb 参数似乎存在错误。代码可能将其解释为字节而不是兆字节，因此请尝试将 32 更改为更大的值。请参阅此处答案中的示例。

于 2015-07-28T12:26:31.903 回答