Я пытаюсь запустить конвейер в Google Cloud DataFlow в "потоковом" режиме. Конвейер должен читать из темы PubSub, однако на самом деле он не читает из темы, пока я не удалю ее, не создаю заново и не опубликую все сообщения в теме ПОСЛЕ запуска конвейера.
Есть ли способ заставить конвейер читать уже опубликованные сообщения?
Пожалуйста, создайте индивидуальную подписку в pub sub с помощью облачной консоли. В коде попробуйте что-нибудь вроде этого.
PCollection<TableRow> datastream = p.apply(PubsubIO.Read.named("Read device iot data from PubSub")
.subscription(String.format("projects/%s/subscriptions/%s",<ProjectId>,<Subscriptionname>))
.timestampLabel("ts")
.withCoder(TableRowJsonCoder.of()));
Обратите внимание, что при подписке вы можете подписаться либо на тему, либо на название подписки.
В приведенном выше коде я подписался на подписку, которую создал явно в суб-консоли pub. Преимущество явной подписки заключается в том, что в ней хранятся данные, полученные из подписки pub, даже когда код потока данных отключен, поэтому данные не будут потеряны.
Похоже на предоставление подписки Pub / Sub (подробнее см. Документация Pub / Sub I / O) решит вашу проблему. Сообщения будут помещены в буфер после создания подписки, что позволит их прочитать при запуске конвейера.
