使用 couchbase-spark-connector 时出错。抛出 BackPressureException

Question

我使用 AWS EMR 和 Spark 1.4.0 作为独立集群运行（不由 Yarn 或 Mesos 管理），我使用 couchbase-spark-connector 版本 1.0.0-beta 作为以及撰写本文时的couchbase-java-client 版本2.2.0-dp2。

我有一个用 Couchbase 编写的视图，它只返回我的文档的键（大约 300M+ 个键）。

我编写了一个用 Scala 2.10.4 编写的 Spark 程序，对于从视图返回的每个键，我正在获取文档并希望将其全部保存到 AWS S3 文件系统。

我的问题是 Couchbase 向我抛出了一个异常类型：com.couchbase.client.core.BackpressureException，documentation:

确定在使用服务时需要在供应商方面退缩，因为消费者超载。

所以我的问题是如何修复这个异常而不被抛出。我想我想延迟我的请求，但我不知道该怎么做。

下面附上我的 Spark 代码：

val couchbaseKeys = sparkContext.couchbaseView(ViewQuery.from(couchbaseDesignName, couchbaseViewName)).map(_.id).couchbaseGet[JsonDocument]()
    couchbaseKeys.map(Projection.projectCouchbaseObject(_)).filter(_ != null).saveAsTextFile(pathForExportedOutput)

Projection 对象及其方法：

object Projection {

  val logger: Logger = LoggerFactory.getLogger(this.getClass)

  def projectCouchbaseObject(couchbaseObject: JsonDocument): String = {
    try {
      return couchbaseObject.id() + '\t' + couchbaseObject.content()
    }
    catch {
      case exception: Throwable => {
        logger.error("Failed project couchbase object. key was: " + couchbaseObject.id(), exception)
      }
    }

    return null
  }
}

Answer 1

尝试调整 ma​​xRetries 和 ma​​xDelay 配置，这样可以解决您的问题。

看这里：https://github.com/couchbase/couchbase-spark-connector/blob/33e4b8382ae80e39ae0254489f5c410f10e7d549/src/main/scala/com/couchbase/spark/connection/CouchbaseConfig.scala#L30