Основная идея Dataproc: как он работает?

Question

Я пытаюсь понять эксплуатационные аспекты dataproc.

Допустим, у меня есть куча CSV-файлов в облачном хранилище, и у меня есть один скрипт Python, который читает их, выполняет некоторые агрегации и сохраняет данные в bigquery.Вот как это работает на одном компьютере.

Если я создам кластер dataproc и разрешу запускать этот сценарий одновременно узлами кластера, как эта вещь будет распараллелена между узлами кластера?Будет ли каждый узел пытаться прочитать все файлы и выполнить агрегации ИЛИ каждый из них автоматически прочитает их соответствующее подмножество?Я просто пытаюсь понять, как это будет работать.Спасибо.

Answer 1

Предположим, что ваши CSV-файлы имеют такую ​​же структуру, а также давайте предположим, что вы написали свой скрипт на Python с использованием Spark API с небольшим преобразованием mapreduce, результат которого будет записан в Bigquery.

1. Будет ли каждый узел пытаться прочитать все файлы и выполнить агрегирование ИЛИ каждый из них автоматически прочитает их соответствующее подмножество?

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

distFile = sc.textFile("file.csv")
distFile = sc.textFile("/hdfs/path/to/folder/fixed_file_name_*.csv")

ПРИМЕЧАНИЕ. Обратите внимание, что существуют дополнительные последствия для расположения файла (ов), это может быть локальный файл (file: //) или удаленный файл (hdfs: //).

Процесс параллельного чтения будет управляться YARN в зависимости от доступности рабочих.Как только эта первая стадия закончена, агрегация (преобразование) может быть выполнена.Этот второй этап также управляется YARN.Предположим, что ваш файл содержит только числовой столбец, тогда это преобразование:

val mapFile = distFile.map(x => x * 2)

переменная mapFile будет содержать файл с тем же числом строк, что и distFile, каждая новая строка является квадратомОригинальный номер.Как видите, вы записываете только преобразование, а YARN будет планировать выполнение, распределяя рабочую нагрузку среди доступных рабочих (подзадачи, выполняющие одну и ту же операцию с разными номерами).

После этого вы будетевозможность записи переменной mapFile в таблицу Bigquery с помощью команды bq load из соединителя Dataproc BQ .

Как эта вещь будет распараллелена между узлами кластера?

Это непростая работа, поскольку необходимо учитывать множество факторов, таких как пространство, память, доступность и т. Д.рабочая сторона, поэтому YARN создана для этого критического планирования решений.На самом деле, существуют разные подходы, которые YARN может использовать для определенных рабочих нагрузок при планировании заданий;CapacitySceduler или FairScheduler.При запуске задания в YARN UI появляется некоторая дополнительная информация.

Answer 2

По сути, при использовании Hadoop MapReduce / Spark в Dataproc для обработки данных в GCS используются 2 уровня абстракций.

Одним из них является уровень файловой системы, соединитель GCS реализует API файловой системы Hadoop, который позволяет пользователям читать / записывать файл из / в GCS, он аналогичен HDFS.Слой файловой системы позволяет выполнять произвольное чтение с любого смещения, но он не знает о формате файла (например, CSV, Parquet, Avro и т. Д.).

Существует еще один слой - InputFormat, который находится сверхууровня файловой системы и знает формат файла.Определенный InputFormat знает, как разбить файл на разбиения (например, разбить файл CSV на несколько разбиений с разными смещениями) и превратить каждое разбиение в записи (например, превратить каждую строку файла CSV в запись).

Когда вы пишете задание MapReduce / Spark, вы знаете формат файла, поэтому выбираете определенный класс InputFormat.Реализация InputFormat может возвращать разбиения (метаданные) файла, затем MapReduce / Spark может распространять разбиения (метаданные) среди разных работников в кластере для параллельной обработки.