У меня задание flink, в котором папка файловой системы nfs является источником, а kafka - приемником. на данный момент преобразований не производится.
Я использовал функцию непрерывного мониторинга для постоянного мониторинга событий в папке и ContinuousFileReaderOperator для чтения данных.
ContinuousFileMonitoringFunction<String> monitoringFunction = new ContinuousFileMonitoringFunction<>(
inputFormat, FileProcessingMode.PROCESS_CONTINUOUSLY, env.getParallelism(),
MONITORING_INTERVAL);
ContinuousFileReaderOperator<String> reader = new ContinuousFileReaderOperator<>(inputFormat);
Начальный размер папки ~ 40 ГБ с 3785468 файлами (во всех подкаталогах) в нем.
Я создал 1 менеджер заданий с кучей 25G и 2 менеджера заданий с 4 слотами заданий и следующими значениями памяти.
taskmanager.memory.process.size: "26g"
taskmanager.memory.flink.size: "24g"
jobmanager.heap.size: "25g"
taskmanager.memory.jvm-overhead.max: "2g"
taskmanager.memory.task.off-heap.size: "1024M"
taskmanager.memory.task.heap.size: "16g"
taskmanager.memory.managed.fraction: 0.2
taskmanager.memory.network.max: "2g"
Когда начатый заданием менеджер заданий работает над подготовкой задания и состояние подготовки занимает много времени около 2 часов. Как только работа запускается, она работает нормально при передаче файлов в kafka.
Я пытаюсь настроить работу, может кто-нибудь помочь мне понять, что происходит на этапе подготовки и какая часть памяти важна в этом состоянии ?
Я пытаюсь поиграть с параметрами памяти, но, похоже, ничего не работает, не зная, какая память используется для чего я не могу продолжить.
Я просмотрел документацию flink по памяти но неясно, для чего используется управляемая память, а для обработки задания используется DirectMemory.
https://ci.apache.org/projects/flink/flink-docs-stable/ops/config.html#memory -конфигурация
Может ли кто-нибудь помочь мне понять, что я должен рассмотреть для точной настройки работы?