Apache Beam Stream от KafkaIO - Нужно окно - PullRequest
Новая
       10

Apache Beam Stream от KafkaIO - Нужно окно

0 голосов
/ 26 апреля 2018

Я передаю сообщения из темы Kafka, используя KafkaIO API https://beam.apache.org/documentation/sdks/javadoc/2.0.0/org/apache/beam/sdk/io/kafka/KafkaIO.html

Трубопроводный поток выглядит следующим образом:

KafkaStream -> Декодирование сообщения с использованием преобразователя -> Сохранить в BigQuery

Я расшифровываю сообщение и сохраняю его в BigQuery с помощью BigQueryIO. Я хотел бы знать, нужно ли мне использовать окно или нет. 

Window.into[Array[Byte]](FixedWindows.of(Duration.standardSeconds(10)))
        .triggering(
          Repeatedly
            .forever(
              AfterProcessingTime
                .pastFirstElementInPane()
                .plusDelayOf(Duration.standardSeconds(10))
            )
        )
        .withAllowedLateness(Duration.standardSeconds(0))
        .discardingFiredPanes()
    )

в соответствии с окном документации требуется в случае, если мы делаем какие-либо вычисления, такие как GroupByKey и т. Д. Поскольку я просто декодирую сообщение массива байтов и сохраняю их в BigQuery, это может не потребоваться.

Пожалуйста, дайте мне знать, мне нужно использовать окно или нет?

1 Ответ

0 голосов
/ 25 мая 2018

Существует ответ, уже отправленный на аналогичный вопрос , где данные передаются из PubSub. Основная идея заключается в том, что невозможно собрать все элементы неограниченной PCollections, поскольку новые элементы постоянно добавляются, и поэтому должна быть реализована одна из двух стратегий:

  • Работа с окнами : сначала вы должны установить неглобальную функцию управления окнами.
  • Триггеры : вы можете настроить триггер для неограниченной коллекции PC таким образом, чтобы он предоставлял периодические обновления для неограниченного набора данных, даже если данные в подписке все еще передаются

Может также потребоваться включить потоковую передачу в конвейере, установив соответствующий параметр arg опции, используя следующую команду: 

pipeline_options.view_as(StandardOptions).streaming = True
...