0

我目前正在使用 Apache Beam 的 scala 包装库 scio。您要做的是根据 ID 组合从 CloudPubSub 发送的不同类型的消息。

消息A每秒发送一次，消息B每三秒发送一次。当我收到消息 B 时，我想将收到的消息 A 中具有相同 ID 的消息合并。

消息示例）

A=37
A=38
A=39
B=39
A=40
A=41
A=42
B=42
A=43
A=44

当前代码

val AInput = sc.pubsubSubscription[String]("projects/hoge/subscriptions/A")`
  .withFixedWindows(Duration.standardSeconds(10))
  .keyBy(a => {
    a.split("=")(1).toInt
  })

val BInput = sc.pubsubSubscription[String]("projects/hoge/subscriptions/B")
  .withFixedWindows(Duration.standardSeconds(10))
  .keyBy(a => {
    println(a.split("=")(1))
    a.split("=")(1).toInt
  })
  .toWindowed
  .map(s => {
    println(s.value.toString)
    println(s.window.maxTimestamp().toDateTime.toString("yyyy/MM/dd HH:mm:ss ZZ"))
    s
  })
  .toSCollection
  .join(AInput)
  .map(a  => {
    println("---------------")
    println(a._1)
    println(a._2._1)
    println(a._2._2)
  })

这两行都执行到 keyBy 行。但是，加入后打印不会打印任何内容。没有错误等...

麻烦。我在等一个答案...

（控制台日志）</p> 

9
3
12
(3,B=3)
(9,B=9)
2017/07/17 16:30:09 +09:00
2017/07/17 16:28:39 +09:00
(12,B=12)
2017/07/17 16:29:09 +09:00
6
9
15
12
(6,B=6)
(9,B=9)
2017/07/17 16:30:19 +09:00
2017/07/17 16:30:09 +09:00
(12,B=12)
2017/07/17 16:30:19 +09:00
(15,B=15)
2017/07/17 16:30:19 +09:00
21
24
27
18
30
(21,B=21)
2017/07/17 16:30:29 +09:00
(24,B=24)
2017/07/17 16:30:39 +09:00
(27,B=27)
2017/07/17 16:30:39 +09:00
(18,B=18)
2017/07/17 16:30:29 +09:00
(30,B=30)
2017/07/17 16:30:39 +09:00
33
36
42
(33,B=33)
2017/07/17 16:30:49 +09:00
39
(42,B=42)
2017/07/17 16:30:59 +09:00
(36,B=36)
2017/07/17 16:30:49 +09:00
(39,B=39)
2017/07/17 16:30:59 +09:00
45

窗口处理似乎每 10 秒进行一次，但处理时间却分崩离析。此外，我发现如果我使用 DataflowRunner 而不是 DirectRunner 启动它，它会成功。

4

0 回答 0