hadoop

Большие данные

1. Статьи о Hadoop

1. Распределенное хранилище Hadoop

1) архитектура хауп

Архитектура Hadoop состоит из трех частей: распределенного хранилища HDFS, распределенных вычислений MapReduce и механизма планирования ресурсов Yarn. Историческая эволюция Hadoop показана на следующем рисунке:

2) Архитектура hdfs

Большинство архитектур больших данных являются архитектурами «главный-подчиненный», а HDFS также является архитектурой «главный-подчиненный» или «узел управления|рабочий узел».

3) Вторичный принцип

1. В чем будет проблема при сохранении исходных данных в памяти NameNode? Какую функцию решить?

При сбое системы исходные данные будут потеряны. Файл журнала редактирования HDFS editlog: в журнале редактирования журнала редактирования в узле NameNode записываются все записи клиента в HDFS. Такие как: добавить, удалить, переименовать и т.д.

Однако размер файла журнала редактирования будет увеличиваться со временем, в результате чего перезапуск системы будет занимать все больше и больше времени для восстановления в соответствии с журналом.Чтобы избежать этой ситуации, hdfs вводит механизм контрольных точек, а пространство имен fsimage является сохранение исходных данных hdfs Контрольная точка — это файл, созданный путем сброса метаданных в памяти на диск.

В это время, если namenode перезапускается, файл fsimage на диске может быть прочитан в содержимое, данные могут быть восстановлены до определенной контрольной точки, а журнал редактирования после контрольной точки выполнен и, наконец, полностью восстановлен. Но все равно с течением времени лог едитлога будет увеличиваться, а при перезапуске namenode будет происходить много событий и hdfs будет недоступен.Чтобы решить эту проблему, hdfs вводит вторичный namenode, который используется для слияния fsimage и editlog.

2, вторичный процесс помещение:

  • Двумя основными условиями создания контрольно-пропускного пункта являются:
    • SecondaryNameNode создает контрольную точку каждый час
    • SecondaryNameNode каждую минуту проверяет, начиная с последней контрольной точки, содержит ли файл журнала редактирования 10 000 транзакций, и если да, то также создает контрольную точку.

Процесс:

  • SecondaryNameNode сначала запрашивает исходный namenode для внесения изменений, чтобы новый журнал редактирования вошел в новый файл.
  • Затем SecondaryNameNode считывает fsimage и редактирует исходный NameNode через HTTP GET.
  • SecondaryNameNode считывает fsimage в память. Затем выполните каждую операцию редактирования. и создайте новый файл fsimage
  • SecondaryNameNode отправляет новый файл fsimage в исходный NameNode через HTTP PUT.
  • Исходный NameNode заменяет старый fsimage новым fsimage, и система обновит файл fsimage, чтобы записать время контрольной точки.
  • После процесса namenode будет иметь последний fsimage и минимальные правки.

4) Принцип хранения в стойке

В компьютерном зале будут стоять стойки, и в каждой стойке будет несколько серверов: Первый блок будет хранить первую копию блока в каталоге HDFS этой машины. Второй блок хранит вторую копию блока на узле данных (d2) в другой стойке (стойке). Третий блок находится на стойке, где находится d2, найдите другой другой узел данных и сохраните третью копию блока. больше реплик, случайные узлы

5) Механизм сердцебиения

Принцип работы:

  • Когда NameNode запустится, в нем запустится сервер ipc.
  • После того, как DataNode запустится, он зарегистрируется в NameNode, и каждые 3 секунды будет отправлять пульс на NameNode.
  • Heartbeat возвращает результаты с NameNode, дающим команды DataNode, такие как копирование данных блока в другой DataNode или удаление блоков данных.
  • Если NameNode не получает пульс от DataNode более 10 минут, узел DataNode считается недоступным.
  • DataNode периодически (6 часов) сообщает текущий отчет о состоянии блока BlockReport на namenode. Отчет о состоянии блока содержит список всех блоков данных в узле данных.

эффект:

  • Инструкции по возврату в DataNode через пульс nameNode
  • Может определить, находится ли DataNode в сети
  • С помощью BlockReport namenode может узнать о ситуации с хранилищем каждого узла данных, например об использовании диска и черном списке. связанные с балансировкой нагрузки
  • При первом запуске кластера hadoop 99,9% блоков не достигли минимального количества реплик, а кластер находился в безопасном режиме с использованием BlockReport.

6) Процесс чтения и записи Hadoop:

7) Процесс написания Hadoop

Подробный процесс:

  • Создайте файл:
    • Клиент HDFS записывает данные в HDFS, сначала вызывает метод DistributedFileSystem.create() и создает новый файл в HDFS.
    • RPC (ClientProtocol.create()) вызывает create() NameNode (NameNodeRpcServer) удаленно, сначала добавьте новый файл по указанному пути в дереве каталогов HDFS.
    • Затем создание нового файла регистрируется в журнале редактирования.
    • После выполнения метода NameNode.create функция DistributedFileSysrem.create() возвращает FSDataOutputStream, который по существу инкапсулирует объект DFSOutputStream.
  • Создайте конвейер потока данных:
    • Клиент вызывает DFSOutputStream.write() для записи данных
    • DFSOutputStream вызывает ClientProtocol.addBlock(), чтобы сначала применить пустой блок данных к NameNode.
    • addBlock() возвращает объект LocationBlock, который содержит информацию о местоположении всех узлов данных текущего блока данных.
    • Создайте конвейер потока данных с информацией о местоположении
  • Запишите данные текущего блока в popeline конвейера потока данных:
    • Когда клиент записывает данные в потоковый конвейер, он сначала записывает данные в чанк контрольного блока размером 512 байт, после его заполнения вычисляется значение контрольной суммы чанка (4 байта).
    • Затем добавьте контрольную сумму к самим данным блока, чтобы сформировать новый блок (516 байт) со значением контрольной суммы.
    • Сохраните в структурированный пакет большего размера, размер пакета составляет 64 КБ.
    • После заполнения пакета он сначала записывается в очередь данных.
    • oascket вынимается из очереди, записывается в конвейер, сначала записывается в datanode1, затем передается из datanode1 в datanode2, а затем передается из datanode2 в datanode3
    • После того, как данные пакета извлечены, они помещаются в ackQueue для ожидания ответа конвейера о подтверждении пакета.
      • Каждый пакет имеет пакет подтверждения подтверждения, а обратный конвейер (dn3->dn2->dn1) возвращает поток данных.
    • Если подтверждение пакета прошло успешно, удалите пакет из ackQueue; в противном случае возьмите пакет из ackQueue, поместите его обратно в dataQueue и повторно отправьте. - Если в файле есть другие блоки для записи после записи текущего блока, продолжайте вызывать метод ackBlock.
    • После того, как данные последнего блока файла будут записаны, снова будет отправлен пустой пакет, указывающий, что блок записан, а затем конвейер будет закрыт.После того, как все блоки будут записаны, close() закрывает поток.
    • ClientProtocol.complete() сообщает namenode, что все блоки в текущем файле завершены.

Отказоустойчивость

  • Узел данных в пайплайне выходит из строя в процессе записи, как восстановить выходной поток
    • Все пакеты, буферизованные ackQueue в потоке данных, будут повторно добавлены в dataQueue.
    • Выходной поток вызывает ClientProtocol.updateBlockForPipeline(), чтобы применить новую временную метку для блока, и namenode запишет новую временную метку.
    • Убедитесь, что даже если неисправный узел данных восстанавливается, временная метка блока на нем не соответствует новой временной метке, записанной узлом имени, затем блок на неисправном узле данных удаляется, а неисправный узел данных удаляется из конвейера.
    • Выходной поток вызывает ClientProtocol.getAdditionalDatanode(), а namenode выделяет новый узел данных для конвейера потока данных и использует новую метку времени для установления конвейера.
    • Узел данных, недавно добавленный в конвейер, еще не сохранил новый блок.Клиент HDFS информирует узел данных в конвейере через DataTransferProtocol о необходимости скопировать блок на новый узел данных.
    • После перестроения конвейера выходной поток вызывает ClientProtocol.updatePipeline() для обновления исходных данных в namenode.
    • После завершения восстановления после сбоя завершается последующая запись

8) Принцип процесса чтения hdfs

Принцип чтения hdfs аналогичен принципу записи данных.

  • Клиент читает файл hdfs, и клиент вызывает метод open объекта DistributedFileSystem объекта файловой системы.
  • Возвращает объект FSDataInputStream.
  • При построении объекта DFSInputStream вызовите метод getBlockLocations namenode, чтобы получить список узлов хранения данных для нескольких блоков в начале файла; список dn для каждого блока будет отсортирован в соответствии с топологией сети, а самые близкие к клиенту будут занимать первое место.
  • Вызовите метод чтения DFSInputStream, сначала прочитайте данные блока 1, установите соединение с ближайшим узлом данных клиента и прочитайте данные.
  • После прочтения закройте поток, установленный с помощью dn
  • Чтение данных следующего блока2
  • После того, как данные файла будут прочитаны, вызовите метод закрытия FSDataInputStream.

Отказоустойчивость:

  • Случай 1: В процессе чтения блока клиент связывается с терминалом datanode.
    • Клиент устанавливает связь со вторым узлом данных, который хранит этот блок, и считывает данные.
    • Зарегистрируйте этот проблемный узел данных, из него не будут считаны данные.
  • Случай 2: клиент читает блок и обнаруживает, что есть проблема с данными блока
    • Когда клиент читает данные блока, он одновременно считывает контрольную сумму блока.Если клиент вычисляет контрольную сумму для прочитанных данных блока, ее значение отличается от прочитанной контрольной суммы, что указывает на то, что данные блока повреждены.
    • Клиент считывает данные блока с других узлов данных, которые хранят копии этого блока (также вычисляются контрольные суммы)
    • В то же время клиент сообщает nodename об этой ситуации

9) HA высокая доступность

Высокая доступность HDFS: Для HDFS NN хранит метаданные в памяти и отвечает за управление пространством имен файловой системы и клиентскими запросами на чтение и запись в HDFS, но есть только одна NN.Как только произойдет единственная точка отказа, вся кластерная система выйдет из строя. Хотя SNN и есть, но это не горячая резервная копия NN. SNN не может быть немедленно переключен на предоставление внешних услуг, то есть HDFS находится в состоянии отключения.

  • hdfs2.x использует архитектуру HA:

    • В кластере HA можно установить два NN, один из которых активен, а другой находится в режиме ожидания.
    • Зоокипер обеспечивает один главный и один резервный
    • NN в активном состоянии отвечает за ответы на все запросы клиентов, а NN в резервном состоянии действует как узел горячего резервирования для обеспечения синхронизации с исходными данными активной NN.
    • При отказе ноды ACTICT кластер zookeeper обнаружит эту ситуацию, и резервная нода товарища немедленно переключится в активное состояние для предоставления внешних сервисов.
    • Убедитесь, что кластер всегда доступен

  • Как сделать горячее резервное копирование данных

    • Standby — это горячее резервирование активной NN, поэтому информация о состоянии активной NN должна синхронизироваться с StandbyNN в режиме реального времени.
    • Синхронизация состояния может быть достигнута с помощью общей системы хранения, такой как NFS (сетевая файловая система), QJM (Quorum Journal Manager) или Zookeeper.
    • Активная NN записывает обновленные данные в общую систему хранения, а резервная NN продолжает следить за системой.Как только обнаруживаются новые данные для записи, она немедленно считывает данные из общедоступной системы хранения и загружает их в собственную память резервной NN, тем самым убедитесь, что он совместим с резервной NN.

  • заблокировать отчет

    • NN обеспечивает связь отображения между блоками данных и фактическими местами хранения.Чтобы добиться быстрого переключения в случае сбоя, необходимо убедиться, что StandbyNN также содержит последнюю связь отображения блоков.
    • Следовательно, необходимо настроить адреса активных и резервных NN для всех dns и отправить информацию о местоположении блока и пульсации на обе NN одновременно.

10) Федерация (понять)

1. Зачем нам федерация? Хотя ha решает проблему древних часов с одной точкой, hdfs по-прежнему имеет проблемы с масштабируемостью, общей производительностью и изоляцией.

  • С точки зрения расширения системы метаданные хранятся в памяти nn, которая ограничена оперативной памятью (каждый файл, каталог, блок занимает около 150 байт)
  • С точки зрения общей производительности, пропускная способность зависит от одного nn.
  • С точки зрения изоляции одна программа может влиять на работу других программ.Если одна программа потребляет слишком много ресурсов, другие программы не могут работать без сбоев.
  • ha по сути является узлом с одним именем

2. Федерация решает вышеперечисленные 3 проблемы

  • В федерации hdfs разработано несколько пространств имен, и каждое пространство имен имеет одно nn или одно основное и одно резервное два nn, так что служба именования hdfs может быть расширена по горизонтали.
  • Эти nns осуществляют свое собственное пространство имен и управление блоками независимо друг от друга и не нуждаются в координации друг с другом.
  • Каждое DN должно регистрироваться во всех NN в кластере и периодически отправлять информацию о пульсе и информацию о блокировке всем NN, чтобы сообщать о своем статусе.
  • Каждая независимая NN федерации HDFS соответствует независимому пространству имен.
  • Каждое пространство имен управляет своим собственным набором блоков, и эти блоки, принадлежащие одному и тому же пространству имен, соответствуют концепции «пула блоков».
  • Каждое DN будет предоставлять блочное хранилище для всех пулов блоков, и каждый блок в пуле блоков фактически хранится в разных DN.

11) Сжатие (понять)

  • Преимущества сжатия файлов:

    • Уменьшить дисковое пространство, используемое данными
    • Ускорьте ввод-вывод данных на диск и в сеть

  • Часто используемые форматы сжатия

    Формат сжатия UNIX-инструменты Алгоритм расширение файла Делимый
    DEFLATE никто DEFLATE .deflate No
    gzip gzip DEFLATE .gz No
    zip zip DEFLATE .zip YES
    bzip bzip2 bzip2 .bz2 YES
    LZO lzop LZO .lzo No
    Snappy никто Snappy .snappy No

  • Класс реализации сжатия Hadoop; оба реализуют интерфейс CompressionCodec

    Формат сжатия Соответствующий кодер/декодер
    DEFLATE org.apache.hadoop.io.compress.DefaultCodec
    gzip org.apache.hadoop.io.compress.GzipCodec
    bzip2 org.apache.hadoop.io.compress.BZip2Codec
    LZO com.hadoop.compression.lzo.LzopCodec
    Snappy org.apache.hadoop.io.compress.SnappyCodec

11) Небольшое управление файлами (понять)

Схема файлов последовательности:

  • Файл SequenceFile, в основном состоящий из записей записей; каждая запись представлена ​​в виде пар ключ-значение.
  • Файлы SequenceFile можно использовать в качестве контейнеров для хранения небольших файлов;
    • Каждая запись сохраняет содержимое небольшого файла
    • Маленькое имя файла используется в качестве ключа текущей записи;
    • Содержимое маленького файла используется как значение текущей записи;
    • Например, 10 000 небольших файлов по 100 КБ, вы можете написать программу для помещения этих файлов в файл SequenceFile.
  • SequenceFile — этоДелимый, поэтому MapReduce может разделить файл на блоки, и каждый блок работает независимо.
  • Конкретная структура (как показано ниже):
    • SequenceFile сначала имеет 4-байтовый заголовок (номер версии файла).
    • затем несколько записей
    • Некоторые маркеры синхронизации точки синхронизации будут случайным образом вставлены между записями, чтобы облегчить позиционирование на границе записи.
  • В отличие от HAR, SequenceFileподдержка сжатия. Структура записи зависит от того, включено ли сжатие
    • Поддерживаются два типа сжатия:
      • Не сжимайте NONE, как показано ниже.
      • Сжать RECORD, как показано ниже
      • Чтобы сжать БЛОК, ① сжимайте несколько записей одновременно; ② в начале каждого нового блока необходимо вставить точку синхронизации, как показано ниже
    • В большинстве случаев сжатие по блокам (примечание: относится к блокам в SequenceFile) является лучшим выбором.
    • Поскольку блок содержит несколько записей, сходство между записями используется для сжатия, и эффективность сжатия выше.
    • Сброс существующих данных в SequenceFile происходит медленнее. Вместо того, чтобы сначала записывать небольшой файл, а затем записывать его в файл SequenceFile, лучше записать данные непосредственно в файл SequenceFile, исключив небольшой файл как посредник.

2. Распределенные вычисления MapReduce

1) Принцип MapReduce

Mapreduce — это распределенная вычислительная среда, основанная на принципе «разделяй и властвуй», состоящая из двух частей: этапа сопоставления (разделенного на небольшие задачи) и этапа сокращения (обобщения результатов небольших файлов).

  • Этап карты (например, количество слов)
    • Предположим, входной файл mr "Унесённые ветром" содержит какой блок: block1, block2, block3.
    • При программировании Mr каждый блок соответствует задаче карты (задаче карты)
    • Возьмем первый пример для анализа:
      • map1 считывает данные блока1 и считывает по одной строке данных блока1 за раз;
      • Сгенерируйте пару ключ-значение (ключ/значение), в качестве входного параметра параметра map(), вызовите метод map()
      • Предполагая, что первая строка прочитана, в качестве ключа (0) используется байтовое смещение лидера текущей строки до начала текущего блока, а в качестве значения используется содержимое текущей строки.
      • В map() разделите содержимое текущей строки значения на пробелы, чтобы получить, какое слово Дорогой | Медведь | Река
      • Превратите слова в пары ключ-значение и выведите (Дорогой, 1) | (Медведь, 1) | (Река, 1); окончательный результат записывается на локальный диск, где находится узел карты
      • После обработки первой строки данных обрабатывается вторая строка, логика такая же, как и выше, когда все данные в блоке обработаны, задача карты завершается.
  • Уменьшить этап
    • Количество задач Reduce задается программой, написанной вами, и логика каждого редукции одинакова, для примера возьмем первую:
      • После завершения задачи map1, редукция1 подключается к карте1 через сеть и получает данные раздела, принадлежащего редукции1, в результате вывода карты1 на сторону редукции1 через сеть.
      • Также подключитесь к map2 и map3, чтобы получить результаты. Наконец, сторона reduce1 получает 4 (dear, 1) пары ключ-значение, поскольку ключи одинаковые, они делятся на одну группу, и 4 (dear, 1) пары ключ-значение преобразуются в [Dear, Iterable( 1, 1, 1] , )], переданные в качестве двух входных параметров для функции reduce()
      • Внутри функции reduce() подсчитайте общее количество дорогих как 4 и выведите (дорогие, 4) в виде пары ключ-значение.
      • Каждая задача сокращения, наконец, выводит файл, и этот файл записывается в HDFS.

Перемешать эскизПеремешать подробную карту

2) перемешивать

Перемешивание — это в основном процесс, в котором выходные данные стороны карты используются в качестве входных данных стороны уменьшения.

  • сторона карты
    • Каждая задача карты имеет соответствующий кольцевой буфер, выходом является пара kv, которая сначала записывается в кольцевой буфер (размер по умолчанию: 100 м). Когда содержимое занимает 80% пространства буфера, фоновый поток помещает буфер в буфер, переполнение данных записывается в файл на диске
    • Во время процесса переполнения задача карты может продолжать записывать данные в кольцевой буфер; однако, если скорость записи выше скорости записи данных переполнения, задача карты приостановит процесс записи данных в кольцевой буфер по истечении 100 минут. окончательно заполнен.Выполнить только процесс записи переполнения и возобновить запись в буфер до тех пор, пока все данные в кольцевом буфере не будут переполнены и записаны на диск
    • Процесс записи переполнения фонового потока на диск состоит из следующих шагов:
      • Сначала разделите каждую переполненную пару kv; количество каждого раздела определяется количеством задач сокращения в программе mr. По умолчанию HashPartitioner используется для вычисления того, к какому разделу принадлежит текущая пара kv. Формула расчета: (ключ. hashCode() и Integer .MAX_VALUE) % numReduceTasks
      • Сортировка в памяти выполняется по ключу пары kv в каждом разделе.
      • Если установлен локальный объединитель агрегации на стороне карты, объединенная операция выполняется над отсортированными данными в каждой секции.
      • Если установлена ​​функция сжатия вывода карты, данные переполнения будут сжаты.
    • Поскольку данные постоянно записываются в кольцевой буфер, переполнение будет срабатывать много раз, и локальный диск в конечном итоге создаст несколько файлов переполнения. Перед завершением задачи карты все файлы переполнения будут объединены в один большой файл переполнения, и это будет разделенный, отсортированный выходной файл (подробности: 1. При объединении файлов переполнения, если имеется не менее 3 файлов переполнения, и если на стороне карты 2. Но если есть только 2 или 1 overflow write files, то операция объединения не сработает, т.к. суть ее в редукции, и виртуальная машина jvm нужно запустить.есть определенные накладные расходы)
  • уменьшить сторону -reduce teak будет получать свои собственные данные раздела (много пар kv) в выходных данных задачи карты через HTTP после завершения каждой задачи карты.

    • Если выходные данные карты относительно малы, они теперь сохраняются в памяти reduce jmv, в противном случае они напрямую записываются на диск reduce.

    • Как только буфер памяти достигает порога (по умолчанию 0,66) или порога количества выходов карты (по умолчанию 1000), запускается слияние, и результат записывается на локальный диск.

    • Если mrprogramming указывает объединение, выполнить операцию объединения при объединении

    • По мере того, как переполнение записывает больше файлов, фоновые потоки объединяют их в более крупные отсортированные файлы.

    • После того, как задача сокращения скопирует все задачи карты, она объединит все файлы переполнения на диске.

    • Объединить 10 за раз по умолчанию

    • Последняя партия объединена, часть данных из памяти, часть из файлов на диске

    • Выполнение слияния, сортировки, групповых этапов

    • Вызовите метод сокращения один раз для каждого набора данных

Общий процесс Shuffle таков: Maptask будет непрерывно собирать пары kv, выдаваемые нашим методом map(), и помещать их в буфер памяти. Когда буфер переполняется (коэффициент по умолчанию равен 0,8), он переполняется на диск. Если Есть много выходных результатов карты, и будет несколько файлов переполнения.Несколько файлов переполнения будут объединены в один большой файл переполнения.В процессе переполнения и слияния файлов необходимо вызвать разделитель для группировки и сортировки по ключу ( по умолчанию). Он заключается в том, чтобы взять модуль числа разделителей в соответствии с хеш-значением ключа), а затем редуцирующая задача отправится на каждую машину с картографической задачей, чтобы получить соответствующие результирующие данные раздела в соответствии с ее собственным номером раздела, и reducetask объединит эти файлы (сортировка слиянием).

После слияния в большой файл процесс перемешивания завершается, а затем вводится логическая операция процесса сокращения (вынуть группу каждой пары ключ-значение из файла и вызвать функцию UDF (определяемый пользователем метод))

Три, планирование пряжи

  • Применение в процессе изготовления пряжи можно разделить на 3 этапа.
    • подача заявки
      • Клиентская программа отправляет приложение в rm и запрашивает экземпляр приложения.
    • Запустите главный экземпляр приложения
      • rm находит nm, который может запускать контейнер, и запускает главный экземпляр приложения в этом контейнере.
      • Мастер приложения регистрируется в rm.После регистрации пользователи могут запрашивать rm для получения информации о своем приложении, а затем они могут напрямую взаимодействовать со своим собственным мастером приложения.
      • appmaster отправляет запрос ресурса в rm в соответствии с запросом ресурса
      • Когда контейнер успешно выделен, мастер приложений запускает контейнер, отправляя сообщение о спецификации запуска контейнера в nm.Сообщение о спецификации запуска контейнера содержит материал, который позволяет контейнеру и мастеру приложения обмениваться данными.
    • Экземпляр мастера приложения управляет выполнением приложения.
      • Приложение запускается в запущенном контейнере в виде задачи и отправляет запущенный процесс, статус и другую информацию мастеру приложения через протокол, специфичный для приложения.
      • Во время работы приложения клиент активно взаимодействует с главным приложением для получения такой информации, как статус выполнения и ход обновления приложения.
      • После выполнения приложения и завершения всей работы appmaster отменяет регистрацию в rm и закрывает его, а все использованные контейнеры также возвращаются в систему

Лайт:

  • 1. Пользователь отправляет заявку в rm
  • 2. rm обращается за ресурсами к мастеру приложений, взаимодействует с определенным nm и запускает первый контейнер для запуска мастера приложений.
  • 3. Мастер приложений и rm регистрируются и обмениваются данными, чтобы подать заявку на ресурсы для выполнения внутренних задач.После получения ресурсов они свяжутся с nm, и соответствующие задачи будут запущены.
  • 4. После завершения всех задач мастер приложения выходит из rm, и все приложение завершает работу.

mapreduce on yarn

  • Отправить задачу

    • Программа упакована в пакет jar, запустите команду hadoop jar на клиенте и отправьте задание в кластер для запуска.
    • Метод submit() задания вызывается в job.waitForCompletion(true), а в этом методе вызывается метод submitJobInternal() объекта JobSubmitter;
      • ②submitClient.getNewJobID() применяется к менеджеру ресурсов для получения идентификатора задания mr.
      • Проверить выходной каталог: сообщить об ошибке, если выходной каталог отсутствует или уже существует
      • Разделение вычислительных задач. Если он не может быть рассчитан, он сообщит об ошибке
      • ③ Задачи, связанные с запуском задания, такие как пакет jar, файл конфигурации и входной фрагмент задания. Он загружается в каталог, названный в соответствии с идентификатором задания в HDFS (копия пакета jar по умолчанию — 10. При выполнении таких задач, как задачи сопоставления и задачи уменьшения, пакет jar можно прочитать из этих 10 копий).
      • ④ Вызовите submitapplication в rm, чтобы отправить задание
    • Клиент запрашивает ход выполнения задания каждую секунду и печатает отчет о ходе выполнения на консоли, если задание изменяется.
    • Если задание выполнено успешно, распечатайте соответствующий счетчик
    • Если задание не выполняется, распечатайте причину сбоя

  • задание инициализации

    • Когда rm получает уведомление о вызове submitApplication(), запрос передается планировщику (планировщику) rm для выделения контейнера.
    • ⑤ rm связывается с указанным nm и информирует nm о запуске контейнера.После того, как nm получает уведомление, он создает контейнер для определенного ресурса
    • ⑤b, а затем запустите appmaster в контейнере
    • ⑥ Мастер приложения должен принять ход выполнения задачи и заполнить отчет, поэтому мастеру приложения необходимо создать несколько объектов учета для записи объектов.
    • ⑦Попросите клиента вычислить входной фрагмент, разделенный на hdfs.
      • Создание задачи сопоставления для каждого разделения сегментов
      • Через значение свойства mapreduce.job.reduces (заданное jog.setNumReduceTasks() во время программирования) узнайте, сколько задач сокращения должно быть создано текущим mr.
      • У каждой задачи есть taskid

  • задание задачи

    • Если это небольшая работа, мастер приложения будет запускать задачу mr в уберизированном виде, а мастер приложения решит заказать задачи mr в jvm.
      • Причина в том, что если каждая задача выполняется на отдельной JVM, JVM нужно запускать отдельно, а для выделения ресурсов требуется время. Задачи в нескольких jvms выполняются параллельно в соответствующих jvms.
      • Было бы эффективнее выполнять все задачи последовательно в аппмастере, тогда бы это делал аппмастер, а таски запускались бы как убер таски
      • Небольшая оценка задачи: менее 10 задач карты, только одна задача уменьшения, размер ввода mr меньше одного размера блока hdfs
      • Перед запуском любой задачи appMaster вызывает метод setupJob(), создает OutputCommitter, создает окончательный каталог вывода задания (обычно каталог в HDFS) и временный каталог вывода задачи (например, каталог вывода промежуточных результатов задание на карту)
    • Если задание не выполняется как задача uber, мастер приложений будет запрашивать контейнер у rm для каждой задачи в задании.
      • Перед тем, как reduce перейдет к этапу сортировки, все задачи карты должны быть выполнены, а контейнеры, к которым должны применяться все задачи карты, должны отдавать приоритет контейнерам, для которых применяется сокращение.
      • После того, как 5% задач карты будут выполнены, к контейнеру будет применено сокращение
      • При подаче заявки на контейнер для задачи карты следите за локализацией данных Планировщик пытается запланировать контейнер на узле, где находится входной осколок задачи карты (мобильные вычисления, а не данные).
      • Задача сокращения может выполняться на любом вычислительном узле в кластере.
      • По умолчанию каждая задача карты, уменьшения назначения 1G памяти, виртуального ядра, решается mapreduce.map.memory.mb, mapreduce.reduce.memory.mb, mapreduce.map.cpu.vcores, mapreduce по свойству reduce.reduce. cpu.vcores

  • задача выполнение задачи

    • Когда планировщик выделяет контейнер nm для текущей задачи и передает эту информацию мастеру приложения; мастер приложения связывается с nm01, чтобы указать nm01 запустить контейнер, и этот контейнер занимает определенное количество ресурсов.
    • После того, как NM01 получит сообщение, запустить контейнер, занимающий указанное количество ресурсов
    • YarnChild запускается в контейнере, а YarnChild выполняет текущую задачу (сопоставление, уменьшение)

  • Ход выполнения задания и обновления статуса

    • Задание задания и каждая из его задач имеют статус (выполняется, успешно завершено, сбой), ход выполнения текущей задачи и счетчик заданий.
    • Во время выполнения задачи ход выполнения и статус (включая счетчик) передаются в appMaster каждые 3 секунды.
    • appMaster суммирует отчетные результаты всех запущенных в данный момент задач.
    • Клиент опрашивает appMaster каждую секунду, чтобы получить последний статус выполнения задания. Если есть какие-либо изменения, они будут напечатаны на консоли.

  • закончить домашнее задание

    • После того, как appMaster получит отчет о завершении последней задачи, установите статус задания на успех
    • Когда клиент опрашивает appMaster, чтобы узнать о ходе выполнения, он обнаруживает, что задание выполнено успешно, и программа выходит из функции waitForCompletion().
    • Вся статистика по заданию выводится в консоль
    • appMaster и контейнер, выполняющий задачу, очищают промежуточные результаты вывода
    • Информация о задании сохраняется сервером истории для будущих пользовательских запросов.

жизненный цикл пряжи

  • RM: Resource Manager
  • AM: Application Master
  • NM: Node Manager
  1. Клиент подает заявку на РМ, включая программу АМ и команду запуска АМ.
  2. RM выделяет первый контейнер для AM и связывается с соответствующим NM, чтобы запустить AM приложения в контейнере.
  3. Когда AM запускается, он регистрируется в RM, позволяя Клиенту получать информацию AM от RM и связываться с AM напрямую.
  4. AM согласовывает ресурсы контейнера для приложений через протокол запроса ресурсов.
  5. Если выделение контейнера прошло успешно, AM требует, чтобы NM запустил приложение в контейнере, и приложение может взаимодействовать с AM независимо после его запуска.
  6. Приложение выполняется в контейнере и отчитывается перед AM.
  7. Во время выполнения приложения клиент и AM взаимодействуют друг с другом для получения состояния приложения.
  8. После завершения выполнения приложения AM выходит из системы и закрывает RM для освобождения ресурсов.

Форматы файлов в Hadoop можно условно разделить на две категории: ориентированные на строки и ориентированные на столбцы:

Ориентированность на строки: данные одной и той же строки хранятся вместе, то есть хранятся непрерывно. SequenceFile, MapFile, Avro Datafile. Таким образом, если требуется доступ только к небольшой части данных строки, вся строка должна быть прочитана в память.Отложенная сериализация может в определенной степени облегчить эту проблему, но накладные расходы на чтение всей строки данных из диска не избежать. Хранилище, ориентированное на строки, подходит для ситуаций, когда необходимо обрабатывать целые строки данных одновременно.

Ориентированность на столбцы: весь файл разбивается на несколько столбцов данных, и каждый столбец данных хранится вместе. Паркет, RCF-файл, ORCF-файл. Формат, ориентированный на столбцы, позволяет пропускать ненужные столбцы при чтении данных и подходит для ситуаций, когда в строке небольшое количество полей. Но операции чтения и записи в этом формате требуют больше места в памяти, потому что кешированная строка должна находиться в памяти (чтобы получить столбец из нескольких строк). В то же время он не подходит для потоковой записи, потому что после сбоя записи текущий файл не может быть восстановлен, а данные, ориентированные на строки, могут быть повторно синхронизированы с последней точкой синхронизации при сбое записи, поэтому Flume принимает строку. -ориентированный формат хранения.

По умолчанию степень сжатия orc storage выше, чем у parquet (формат сжатия также можно изменить. При том же формате сжатия, поскольку схема данных формата parquet более сложная, занимаемое пространство немного выше. формат сжатия, орк может достигать 1:3 Коэффициент сжатия выше, паркет чуть ниже 1:3); Вообще говоря, эффективность чтения orc выше, чем у паркета (родной формат хранения для hadoop, более удобный для поддержки ульев); Parquet поддерживает вложенные форматы данных, orc изначально не поддерживает вложенные типы данных (но это может быть реализовано косвенно через сложные типы данных, такие как map, где ситуация соответствует «особому случаю» в статье 2, которая затрагивает представление); Parquet поддерживает расширение поля, orc не поддерживает расширение поля нативно (но реализацию метода чтения можно переписать вручную, что также соответствует "частному случаю" во второй статье);

Подать заявку на ресурсы->Запустить appMaster->Подать заявку на контейнер для выполнения задач->Распределить задачу->Выполнить задачу->Завершение задачи->Перезапустить контейнер

Категории