沙发底座火花连接器 DCP 从最后位置恢复

Question

我有这个 spark 应用程序：

 val conf = new SparkConf().setMaster("local[*]")
                           .setAppName("StreamingSample")
                           .set("com.couchbase.bucket.test", "")
                           .set("com.couchbase.nodes", "test-machine")
 val ssc = new StreamingContext(conf, Seconds(5))

 ssc.couchbaseStream(from = FromNow, to = ToInfinity)
    .filter(!_.isInstanceOf[Snapshot]) // Don't print snapshots, just mutations and deletions
    .checkpoint(Seconds(2))
    .foreachRDD(rdd => {
  val om: Broadcast[ObjectMapper] = ScalaObjectMapper.getInstance(rdd.sparkContext)

  rdd.foreach {
    case m: Mutation =>

      val content: Map[String, Object] = om.value.readValue(m.content, classOf[Map[String, Object]])

      content("objectType") match {
        case "o" => println("o")
        case "c" => println("c")
        case "s" => println("s")
        case unsupportedType => println("unsupported")
      }


    case m: Deletion => println("delete")
  }

})

当 recover spark 失败时，我如何从上一个位置恢复？

Answer 1

遗憾的是，当前的连接器版本 (1.2.1) 只能从开头或从当前位置（流的末尾）流式传输。因此，在您的示例中，您别无选择，只能将 FromNow 更改为 FromBeginning，然后（在代码中）跳过您已经看到的所有消息，直到您赶上为止。

客户端团队目前正在开发一种能够记住状态的新实现，因此您将能够从流中的特定点恢复。

沙发底座火花连接器 DCP 从最后位置恢复

couchbase spark connector DCP recover from last position

couchbase

apache-spark