Beam Data Flow записывает данные только в temp в GCS

Question

У меня есть очень простое задание Python Dataflow, которое считывает некоторые данные из Pub / Sub, применяет FixedWindow и записывает в Google Cloud Storage.

transformed = ...
transformed | beam.io.WriteToText(known_args.output)

Вывод записывается в определенное место в -вывод, но только временная стадия, то есть

gs://MY_BUCKET/MY_DIR/beam-temp-2a5c0e1eec1c11e8b98342010a800004/...some_UUID...

Файл никогда не помещается в правильно названное место с шаблоном шардинга.

Проверено на локальном и DataFlow Runner.

---

При дальнейшем тестировании я заметил, что в примере streaming_wordcount есть те же проблемы, однако стандартный пример wordcount пишет нормально.Возможно, проблема связана с управлением окнами или чтением из pubsub?

---

Похоже, WriteToText не совместим с потоковым источником PubSub.Вероятно, есть обходные пути, или версия Java может быть совместимой, но я решил использовать другое решение в целом.

Answer 1

Экспериментальное выполнение потокового конвейера Python доступно (с некоторыми ограничениями).

Неподдерживаемые функции применяются ко всем участникам.API-интерфейсы состояний и таймеров, API-интерфейс пользовательских источников, API-интерфейс Splittable DoFn, Обработка поздних данных, Пользовательские пользовательские WindowFn .

Кроме того, DataflowRunner в настоящее время не поддерживаетследующие специфические функции облачного потока данных при выполнении потоковой передачи Python.

автоматическое масштабирование потока Обновление существующих конвейеров. Шаблоны потока данных облака. Некоторые функции мониторинга, такие как счетчики мс, отображаемые данные, метрики и количество элементов для преобразований.Однако поддерживаются журналы, водяные знаки и количество элементов для источников.

https://beam.apache.org/documentation/sdks/python-streaming/

Поскольку вы используете FixedWindowFn, и конвейер смог записать данные в местоположение tmp, перепроверьтевыходное местоположение --output gs://<your-gcs-bucket>/<you-gcs-folder>/<your-gcs-output-filename>