В любом случае можно ли преобразование набора данных: «Distinct ()» использовать в Datastream во Flink?

Question

Мне было интересно, если в любом случае будет использоваться API потока данных Flink для удаления дубликатов из поступающих записей (может быть в течение определенного временного окна), так же, как в API набора данных, который обеспечивает преобразование под названием «Distinct». Или в любом случае, если набор данных можно преобразовать в поток данных, учитывая, что набор данных преобразуется в поток данных для внутренней обработки в Flink.

Пожалуйста, помогите мне в этом. Заранее спасибо! Ура!

Answer 1

Я не знаю ни о каком встроенном примитиве, но если все данные в окне помещаются в память, вы можете легко построить эту функцию самостоятельно.

DataStream<...> stream = ...
stream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
    .process(new DistinctFunction<>());

public class DistinctFunction<T, W extends Window> extends ProcessAllWindowFunction<T, T, W> implements Function {
    public void process(final Context context, Iterable<T> input, Collector<R> out) throws Exception {
        Set<T> elements = new HashSet<>();
        input.forEach(elements::add);
        elements.forEach(out::collect);
    }
}

Конечно, это намного больше Масштабируемость, если у вас есть ключ, так как только данные одного ключа в окне должны храниться в памяти.