使用 SCIO来自 spotify
为 Dataflow
写一份工作, 以下 2 个例子 e.g1和 e.g2写一个 PubSub
流到 GCS
,但以下代码出现以下错误
错误
Exception in thread "main" java.lang.IllegalArgumentException: Write can only be applied to a Bounded PCollection
代码
object StreamingPubSub {
def main(cmdlineArgs: Array[String]): Unit = {
// set up example wiring
val (opts, args) = ScioContext.parseArguments[ExampleOptions](cmdlineArgs)
val dataflowUtils = new DataflowExampleUtils(opts)
dataflowUtils.setup()
val sc = ScioContext(opts)
sc.pubsubTopic(opts.getPubsubTopic)
.timestampBy {
_ => new Instant(System.currentTimeMillis() - (scala.math.random * RAND_RANGE).toLong)
}
.withFixedWindows((Duration.standardHours(1)))
.groupBy(_ => Unit)
.toWindowed
.toSCollection
.saveAsTextFile(args("output"))
val result = sc.close()
// CTRL-C to cancel the streaming pipeline
dataflowUtils.waitToFinish(result.internal)
}
}
我可能将窗口概念与 Bounded PCollection 混淆,有没有办法实现这一点,或者我是否需要应用一些转换来实现这一点,任何人都可以提供帮助
最佳答案
我相信SCIO的saveAsTextFile
下面使用 Dataflow 的 Write
转换,仅支持有界 PCollections。 Dataflow 尚未提供直接 API 来将无界 PCollection 写入 Google Cloud Storage,尽管这是我们正在研究的问题。
要将无界 PCollection 保留在某处,请考虑例如 BigQuery、Datastore 或 Bigtable。例如,在 SCIO 的 API 中,您可以使用 saveAsBigQuery
.
关于scala - 使用 Dataflow 将 PubSub 流写入 Cloud Storage 时出错,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/39865774/