1. Концепция Flink State

Состояние используется для записи промежуточных результатов вычислений или метаданных операторов во время работы приложений Flink. Если работающему приложению Flink для обработки требуется последний результат расчета, ему необходимо использовать состояние для хранения промежуточного результата расчета. Например, сценарии объединения и агрегирования окон.

Когда приложение Flink запущено, информация о состоянии будет сохранена в экземпляре объекта State. Экземпляр объекта State сохраняет соответствующие данные в файловой системе FS или базе данных RocksDB через StateBackend. Пока приложение Flink работает, данные о состоянии периодически сохраняются с помощью снимков контрольных точек. А когда приложение Flink перезапускается, загружается контрольная точка/точка сохранения для восстановления состояния, чтобы приложение Flink могло продолжить выполнение предыдущего расчета данных и обеспечить точную передачу данных в нисходящий поток.

1.1 Реализация хранилища состояний в классификации Apache Flink StateBackend

Разделены на следующие 3 категории:

• HeapStateBackend на основе памяти. Состояние сохраняется в памяти.
• FsStateBackend на основе HDFS или OSS. Состояние сохраняется в памяти и сохраняется на удаленном конце при выполнении cp (контрольная точка).
• RocksDBStateBackend на основе RocksDB. Сериализуйте объект в двоичный формат и сохраните его в данных RocksDB в памяти и на локальном диске, а затем сохраните его на удаленном конце во время cp.

HeapStateBackend и RocksDBStateBackend соответственно соответствуют своим позициям в модели памяти TaskManager:

Структура хранения в RocksDBStateBackend:

пространство имен: в разных пространствах имен существуют состояния с одинаковым именем.

1.1.1 Сохранение состояния состояния

проходить Chandy-Lamport алгоритм распределенных снимков checkpoint Завершите сохранение данных состояния. затем в Flink Читать, когда приложение перезапускается State Данные о состоянии можно восстановить на месте эксплуатации.

Категория КПП:

• Полная контрольная точка на основе памяти
• Контрольная точка полной суммы HDFS
• Полная контрольная точка RocksDB/инкрементная контрольная точка

1.2 Государственная классификация на основе операторов и группировки данных

Состояние можно разделить на две категории: состояние оператора и состояние с ключом.

• Состояние оператора (называемое состоянием без ключа)

Часто существует в Source, Sink. Пример конкретного класса реализации: BroadcastState

Пример: ОператорState используется для записи смещения в Kafka Source.

• Keyed State

Любой тип ключевого состояния может иметь период действия (TTL), и все типы состояний поддерживают одноэлементный TTL. Это означает, что срок действия элементов List и Map истекает независимо.

Пример: данные в окне после SQL GroupBy/PartitionBy, каждый ключ имеет соответствующее состояние. Данные о состоянии между ключом и ключом не видны.

Конкретный класс реализации ключевого состояния:

• ValueState
• MapState
• ListState
• AggregatingState
• ReducingState
• 。。。。。

Интеллект-карта штата Флинк:

2. Общие процедуры обработки, связанные со статусом

2.1 Как сохраняется состояние в приложениях Flink?

1. Как Kafka Source хранит OperationState?

class FlinkKafkaConsumerBase {
 private transient ListState<Tuple2<KafkaTopicPartition, Long>> unionOffsetStates; // имя состояния: "topic-partition-offset-states"
// Особый тип государства: Союз State 
}

Переменная UnionOffsetStates имеет тип OperationState.

1. Как оператор Map хранит данные, которые необходимо накопить?

• ValueState/MapState/ListState/......

думать：keyby Распределение постданных и несколько степеней параллелизма subtask Какова связь между ними?

Во-первых, после того, как данные в потоке данных пройдут через Keyby, они будут разделены на различные KeyedStreams. Каждый KeyedStream имеет свой собственный KeyedState (например, ValueState/ListState/MapState).

Во-вторых, данные в KeyedStream будут организованы в KeyGroup. KeyGroup — это наименьшая единица Flink для перераспределения состояния ключей.

Наконец, группа ключей серединаиз Данные будут распределяться по каждому модулю с использованием максимальной степени параллелизма по модулю. subtask середина. Ниже приведен ключевой исходный код:

KeyGroupStreamPartitioner#selectChannel(record)
{
    K key;
    key = keySelector.getKey(record.getInstance().getValue());
    return KeyGroupRangeAssignment.assignKeyToParallelOperator(
            key, maxParallelism, numberOfChannels);
}
--KeyGroupRangeAssignment#assignKeyToParallelOperator（）
    {
    return computeOperatorIndexForKeyGroup(maxParallelism, parallelism, assignToKeyGroup(key, maxParallelism));
    }
    --KeyGroupRangeAssignment#computeOperatorIndexForKeyGroup（）
      Официально: Оператор Индекс = keyGroupId * parallelism / maxParallelism
    --KeyGroupRangeAssignment#assignToKeyGroup()
      {
        return computeKeyGroupForKeyHash(key.hashCode(), maxParallelism);
       }

2.2 Изменения в хранилище состояний при изменении сценария параллелизма

2.3 Отношения между государством и контрольно-пропускным пунктом

Распределенный снимок Checkpoint концепция, регулярно State упорствовать в Внешняя система хранения (HDFS/OSS) начальство. Пользователи могут провестивыполнение CheckpointedFunction интерфейс для использования operator state。проходить barrier выравнивать КПП, подожди State Персистентность завершена (этот процесс также может быть асинхронным в зависимости от параметров).

общий State и CP Связанные вопросы：

• Штат Штат слишком велик. Проблема заключается в том, что контрольные точки нескольких операторов или нескольких подзадач одного оператора выполняются медленно, что может привести к длительному времени выравнивания CP, а в тяжелых случаях – к тайм-ауту CP.
• Неравномерность данных приводит к несвоевременной обработке определенной подзадачи. Проблема заключается в том, что контрольные точки нескольких подзадач одного оператора выполняются медленно, что приводит к длительному времени выравнивания CP. В тяжелых случаях может произойти тайм-аут CP.
• Большие задания (параллелизм) часто выполняют CP и часто загружают небольшие файлы, что приводит к появлению слишком большого количества маленьких файлов в кластере HDFS.

Распространенные решения: увеличьте размер управляемой памяти.

3. Справочные документы:

• Flink State Официальная документация：Flink состояние и отказоустойчивость
• https://cloud.tencent.com/developer/article/1403939
• https://www.modb.pro/db/81206