поток BigQuery таблицы в Google Pub/Sub

Question

У меня есть Google BigQuery Таблица и я хочу, чтобы поток весь стол в паб к югу Тема

, что должно быть простой/быстрый способ сделать это?

Спасибо заранее,

Answer 1

Это действительно зависит от размера таблицы.

Если это небольшая таблица (несколько тысяч записей, несколько столбцов), вы можете настроить процесс на запрос всей таблицы, преобразовать ответ в массив JSON и нажать на pub-sub.

Если это большой стол (миллионы/миллиарды записей, сотни столбцов) вы должны экспортировать в файл, а затем подготовить/корабль паб к югу

Это также зависит от политики разделения - если ваши таблицы настроены на разделение по дате, вы можете снова запросить вместо экспорта.

Последнее, но не менее важное: это также зависит от частоты - это одноразовая сделка (затем экспорт) или непрерывный процесс (затем используйте декораторы таблицы для запроса только последних данных)?

Нужна дополнительная информация, если вы хотите по-настоящему полезный ответ.

Редактировать

на основе ваших комментариев по размеру стола, я думаю, что лучше всего было бы иметь скрипт, который будет:

1. Экспорт таблицы в GCS в качестве символа новой строки разделителями JSON

2. Обработать файл (строка за строкой чтения) и отправить в паб-суб

Для большинства языков программирования есть client libraries. Я сделал аналогичные вещи с Python, и это довольно прямолинейно.

Answer 2

Самый простой способ, которым я знаю, - это использовать Google Cloud Dataflow, который изначально знает, как получить доступ к BigQuery и Pub/Sub.

В теории это должно быть так же просто, как следующие строки Python:

p = beam.Pipeline(options=pipeline_options) 
tablerows = p | 'read' >> beam.io.Read(
    beam.io.BigQuerySource('clouddataflow-readonly:samples.weather_stations')) 
tablerows | 'write' >> beam.io.Write(
    beam.io.PubSubSink('projects/fh-dataflow/topics/bq2pubsub-topic')) 

Эта комбинация Python/Dataflow/BigQuery/PubSub не работает сегодня (Python DataFlow находится в бета-версии, но keep an eye on the changelog).

Мы можем сделать то же самое с Java, и он хорошо работает - я просто протестировал его.Она работает локально, а также в размещаемого Dataflow бегуна:

Pipeline p = Pipeline.create(PipelineOptionsFactory.fromArgs(args).withValidation().create()); 

PCollection<TableRow> weatherData = p.apply(
     BigQueryIO.Read.named("ReadWeatherStations").from("clouddataflow-readonly:samples.weather_stations")); 
weatherData.apply(ParDo.named("tableRow2string").of(new DoFn<TableRow, String>() { 
    @Override 
    public void processElement(DoFn<TableRow, String>.ProcessContext c) throws Exception { 
     c.output(c.element().toString()); 
    } 
})).apply(PubsubIO.Write.named("WriteToPubsub").topic("projects/myproject/topics/bq2pubsub-topic")); 

p.run(); 

Test, если сообщения есть с:

gcloud --project myproject beta pubsub subscriptions pull --auto-ack sub1 

---

Hosted Dataflow скриншот: