java - 使用 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 编写的 View ，它只返回我文档的 key (大约 3 亿多个 key )。

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

我的问题是 Couchbase 向我抛出一个类型异常:com.couchbase.client.core.BackPressureException，其表示为 documentation :

Identifies the need to back off on the supplier side when using a service, because the consumer is overloaded.

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

下面附有我的 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