внешний вызов API в потоке данных Apache Beam

Если вам нужно проверять дубликаты во внешнем хранилище для каждой записи JSON, вы все равно можете использовать для этого DoFn. Есть несколько аннотаций, таких как @Setup, @StartBundle, @FinishBundle и т. Д., Которые можно использовать для аннотирования методов в вашем DoFn.

Например, если вам нужно создать экземпляр клиентского объекта для отправки запросов во внешнюю базу данных, вы можете сделать это в @Setup методе (например, конструкторе POJO), а затем использовать этот клиентский объект в своем @ProcessElement методе.

Рассмотрим простой пример:

static class MyDoFn extends DoFn<Record, Record> {

    static transient MyClient client;

    @Setup
    public void setup() {
        client = new MyClient("host");
    }

    @ProcessElement
    public void processElement(ProcessContext c) {
        // process your records
        Record r = c.element();
        // check record ID for duplicates
        if (!client.isRecordExist(r.id()) {
            c.output(r);
        }
    }

    @Teardown
    public void teardown() {
        if (client != null) {
            client.close();
            client = null;
        }
    }
}

Кроме того, чтобы избежать удаленных вызовов для каждой записи, вы можете группировать записи в пакетном режиме во внутренний буфер (входные данные Beam разделяются на пакеты) и проверять дубликаты в пакетном режиме (если ваш клиент поддерживает это). Для этой цели вы можете использовать аннотированные методы @StartBundle и @FinishBundle, которые будут вызываться непосредственно до и после обработки пакета Beam соответственно.

Для более сложных примеров я бы рекомендовал взглянуть на реализации Sink в разных IO Beam, например KinesisIO ​​, например.

Пример пакетного вызова внешней системы с использованием DoFn с отслеживанием состояния приведен в следующем сообщении блога: https://beam.apache.org/blog/2017/08/28/timely-processing.html, может быть полезным.