Поток данных / Apache Beam - Как спроектировать конвейерные действия по порядку? - PullRequest
Новая
       65

Поток данных / Apache Beam - Как спроектировать конвейерные действия по порядку?

0 голосов
/ 03 октября 2018

Как часть потока данных / Apache Beam, я хотел бы прочитать из источника, затем записать в источник, а затем снова прочитать из одного источника, а затем снова записать в этом порядке.Как я могу обеспечить порядок R-> W-> R-> W ниже?Я полагаю, что ниже работает как параллельный конвейер с R-> W.Я не уверен, что для этого нужно использовать объект PDone.

(В приведенном ниже примере рассмотрим, что BIGQUERYVIEWB - это представление Big Query, сформированное из TESTDATASET1.TABLE2 и нескольких других таблиц) 

//Read 1
PCollection<TableRow> tr = pipeline.apply(BigQueryIO.readTableRows().fromQuery("SELECT ID FROM `TESTDATASET1.BIGQUERYVIEWA`").usingStandardSql());
PCollection<TableRow> tr1= tr.apply(ParDo.of(new SomeFn()));
//Write 1
tr1.apply(BigQueryIO.writeTableRows().withoutValidation()
                    .withSchema(FormatRemindersFn.getSchema())
                    .withWriteDisposition(WriteDisposition.WRITE_APPEND)
                    .to("TESTDATASET1.TABLE2"));
//Read 2
PCollection<TableRow> tr2 = pipeline.apply(BigQueryIO.readTableRows().fromQuery("SELECT ID FROM `TESTDATASET1.BIGQUERYVIEWB`").usingStandardSql());
PCollection<TableRow> tr3= tr.apply(ParDo.of(new SomeFn()));
//Write 2
tr3.apply(BigQueryIO.writeTableRows().withoutValidation()
                    .withSchema(FormatRemindersFn.getSchema())
                    .withWriteDisposition(WriteDisposition.WRITE_APPEND)
                    .to("TESTDATASET1.TABLE3"));

1 Ответ

0 голосов
/ 05 октября 2018

Посмотрите на преобразование Wait.On (), позволяющее настроить зависимые шаги.

Вот простой пример псевдокода: 

PCollection<String> data = ...;

PCollection<Something> first = data.apply(ParDo.of(..))

data.apply(Wait.On(first)).apply(ParDo.of(..))

Обратите внимание, что Wait.On(..) требует PCollection в качестве сигнала, а не PDone.Я считаю, что BigQuery.Write возвращает WriteResult, из которого вы можете извлечь PCollection для неудачных вставок.

...