无法使用火花结构化流计算文档

Question

我正在尝试使用 couchbase 作为使用 spark 连接器的 spark 结构化流的流源。

val records = spark.readStream
.format(“com.couchbase.spark.sql”).schema(schema)
.load()

我有这个问题

records
.groupBy(“type”)
.count()
.writeStream
.outputMode(“complete”)
.format(“console”)
.start()
.awaitTermination()

对于这个查询，我没有得到正确的输出。我的查询输出表是这样的

Batch: 0
20/04/14 14:28:00 INFO CodeGenerator: Code generated in 10.538654 ms
20/04/14 14:28:00 INFO WriteToDataSourceV2Exec: Data source writer org.apache.spark.sql.execution.streaming.sources.MicroBatchWriter@17fe0ec7 committed.
±-------±----+
|type | count|
±-------±----+
±-------±----+

但是，如果我使用沙发底座将文档作为非流式获取。喜欢

val cdr = spark.read.couchbase(EqualTo(“type”, “cdr”))
cdr.count()

已正确推断此非流式操作的架构，并为结构化流式传输使用相同的架构。

INFO N1QLRelation: Inferred schema is StructType(StructField(META_ID,StringType,true), StructField(_class,StringType,true), StructField(accountId,StringType,true), 

给出正确的输出。 （计数 = 28）。

请告诉我为什么这不适用于结构化流媒体。

Answer 1

这可能是因为您只流式传输从现在开始发生的变化，而不是过去的事件。 如果您想“从头开始”流式传输所有内容，则需要指定。

此博客文章中显示了示例：https://blog.couchbase.com/couchbase-spark-connector-2-0-0-released/

基本上在您的流中，您需要指定以下行

  .couchbaseStream(from = FromBeginning, to = ToInfinity)