ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); ...... JobExecutionResult jobExecutionResult = env.execute(XXXOffline.class.getName()); int records =...

Я хотел бы сохранить все необработанные строки для каждого ключа в запросе Select .. From .. GROUP BY .. во Flink. Я определил AggregateFunction под названием RowToJsonAgg, который объединяет строки в строку Json. class RowToJsonAgg extends...

Я пытаюсь понять, что делает Flinks Collector.collect () и как он обрабатывает входящие / исходящие данные: Пример взят из Flink DataSet API: Следующий код преобразует DataSet текстовых строк в DataSet слов: DataSet<String> output =...

Мое требование - обработать или построить некоторую логику вокруг результата запроса sql во flink. Для простоты скажем, что у меня есть два sql-запроса, которые они выполняют с разным размером окна и один поток событий. У меня вопрос а) как я...

Я хочу отсортировать POJO DataSet на основе нескольких значений с помощью нескольких функций KeySelector: DataSet<PoJo> data = input .sortPartition(new KeySelector<PoJo, Integer>() { public Integer...

Набор данных flink 1.7.2 не поддерживает приемник кафка? После выполнения пакетной операции мне нужно опубликовать сообщение в kafka, то есть источник - мой postgres, а приемник - моя kafka. Является ли это возможным ?

Вот требование: набор данных слишком велик, нам нужно разделить данные, вычислить локальный результат в каждом разделе, а затем объединить. Например, если имеется 1 миллион фрагментов данных, разделенных на 100 разделов, каждая копия будет содержать...

Поддерживает ли Flink функцию бокового вывода в наборе данных (пакетный API)? Если нет, как обрабатывать действительные и недействительные записи при загрузке из файла?

У меня есть задание flink (2 диспетчера задач) с параллелизмом заданий 64 и слотом задач 64. У меня для одного из операторов установлено значение параллелизма 16. Слоты этого оператора (16 параллелизма) не распределяются равномерно между двумя...

Я хотел бы предоставить конечную точку из моего потокового приложения flink. Которое возвращает некоторые статические метаданные о приложении. Каковы возможные пути реализации этого. Пожалуйста помоги

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

У нас есть список путей к файлам из таблицы БД с отметкой времени ее создания. Попытка выяснить, как мы можем использовать список путей к файлам из db для пересылки только тех файлов из nfs в приемник kafka. Прямо сейчас я использую настраиваемую...

Я пытаюсь настроить сеанс flink-yarn для запуска более 100 пакетных заданий. После подключения к ~ 40 диспетчерам задач и ~ 10 запущенным заданиям (каждый диспетчер задач с 2 слотами и 1 ГБ памяти каждый), похоже, сеанс становится нестабильным....

Сведения о кластере Flink, Количество узлов : 4 Версия Flink : 1.11 Клиент Flink : RestCluserClient Мы отправляем пакетное задание Flink из потокового задания с помощью PackagedProgram, но наше требование состоит в том, чтобы выполнять...

Я пытаюсь рассчитать индекс отношений AdamicAdar в следующем графике социальных сетей. Я настроил свои ребра, вершины, набор данных и график, используя библиотеку apache flink-gelly. Вот мой код: import...