Обзор ADB Spark Connector

Михаил Серов

ADB Spark Connector предоставляет возможность высокоскоростного параллельного обмена данными между Apache Spark и Arenadata DB.

ПРИМЕЧАНИЕ

Начиная с версии ADH 4.3.0 ADB Spark Connector поддерживает ADB 7 (Greengage DB 7). Также имя коннектора было изменено с adb на greengage, в связи с чем рекомендуется использовать новое имя, как показано ниже:

# Рекомендуемое имя коннектора, начиная с 4.3.0:
df.write.format("greengage")...

# Устаревшее имя коннектора для обратной совместимости:
df.write.format("adb")...

Старое значение format("adb") поддерживается для обратной совместимости.

Архитектура

Каждое приложение Spark состоит из контролирующего процесса (driver) и некоторого количества распределённых worker-процессов (executors). Схема взаимодействия представлена ниже.

Архитектура ADB Spark Connector
Архитектура ADB Spark Connector
Архитектура ADB Spark Connector
Архитектура ADB Spark Connector

Алгоритмы работы

Чтение данных

ADB Spark Connector инициализирует драйвер, чтобы загрузить таблицу ADB в Spark. Этот драйвер устанавливает соединение с ADB master через JDBC, чтобы получить необходимые метаданные о таблице. Метаданные содержат информацию о типе, структурах таблиц и ключе дистрибуции. Ключ дистрибуции позволяет вам распределить данные из таблицы на существующие ноды Spark.

Данные хранятся в сегментах ADB. Приложение Spark, которое использует коннектор, может перемащать данные из каждого сегмента в одну или несколько партиций Spark. Существует пять режимов партиционирования:

  • В соответствии с gp_segment_id. Данные из таблиц ADB прочитываются и распределяются между партициями Spark в зависимости от сегмента. Количество партиций Spark соответствует количеству активных сегментов ADB. Этот режим партиционирования не требует дополнительных параметров и используется по умолчанию.

  • В соответствии с указанными столбцом и количеством партиций. Данные распределяются между партициями Spark внутри столбцевого диапазона согласно указанному количеству партиций. Количество партиций Spark соответствует количеству рядов. Необходимо установить подходящие параметры (см. Опции ADB Spark Connector). У этого режима партиционирования есть ограничения: в качестве столбца могут использоваться только поля таблицы типа integer или date/time.

  • В соответствии с указанным столбцом. Данные распределяются между партициями Spark в зависимости от уникальных значений в указанном столбце. Количество партиций Spark соответствует количеству уникальных значений в указанном столбце. Необходимо установить подходящие параметры (см. Опции ADB Spark Connector). Этот режим рекомендуется для случая малого и ограниченного набора значений.

  • В соответствии с указанным количеством партиций. Данные распределяются между партициями Spark в зависимости от указанного количества партиций. Необходимо установить подходящие параметры (см. Опции ADB Spark Connector).

  • В соответствии с указанными хеш-функцией и количеством партиций. Данные распределяются между партициями Spark в зависимости от указанной хеш-функции и указанного количества партиций. Необходимо указать подходящие параметры (см. Опции ADB Spark Connector). У этого режима партиционирования есть ограничения: только выражение, возвращающее целое число, может использоваться в качестве хеш-функции.

Некоторые задачи обработки данных распределяются между всеми executor-нодами Spark вместе с соответствующими партициями, которые были созданы ранее. Обмен данными выполняется в параллельном режиме с использованием внешних writable-таблиц для каждой партиции.

Алгоритм чтения данных
Алгоритм чтения данных
Алгоритм чтения данных
Алгоритм чтения данных

Запись данных

ADB Spark Connector инициализирует драйвер для загрузки таблицы из Spark в ADB. После этого драйвер устанавливает соединение с ADB master через JDBC и готовит загрузку ADB. Метод загрузки зависит от режима записи.

Доступны следующие режимы записи:

  • overwrite. В зависимости от spark.db.table.truncate целевая таблица либо удаляется полностью, либо все данные очищаются (cleaned up) перед процессом записи.

  • append. Дополнительные данные записываются в целевую таблицу.

  • errorIfExists. Если целевая таблица существует, появляется ошибка.

Также доступно несколько опций для spark.db.create.table. Это позволяет определять дополнительные настройки, когда вы создаёте целевую таблицу (см. Опции ADB Spark Connector).

Задача загрузки данных создаётся для каждой партиции Spark. Обмен данными происходит параллельно через внешние readable-таблицы для каждого сегмента.

Алгоритм записи данных
Алгоритм записи данных
Алгоритм записи данных
Алгоритм записи данных

Поддерживаемые типы данных

Соответствия типов данных при их перемещениях представлены в таблицах ниже.

Перемещение из ADB в Spark
ADB Spark

bigint

LongType

bigSerial

LongType

bit

StringType

bytea

BinaryType

boolean

BooleanType

char

StringType

date

DateType

decimal

DecimalType

float4

FloatType

float8

DoubleType

int

IntegerType

interval

CalendarIntervalType

serial

IntegerType

smallInt

ShortType

text

StringType

time

TimeStampType

timestamp

TimeStampType

timestamptz

TimeStampType

timetz

TimeStampType

varchar

StringType
Перемещение из Spark в ADB
Spark ADB

BinaryType

byte

BooleanType

boolean

CalendarIntervalType

interval

DateType

date

DecimalType

numeric

DoubleType

float8

FloatType

float4

IntegerType

int

LongType

bigint

ShortType

smallint

StringType

text

TimeStampType

timestamp

Поддерживаемые драйверы JDBC

Для взаимодействия ADB Spark Connector и ADB используется JDBC-соединение. Драйвер PostgreSQL JDBC driver поставляется вместе с коннектором.

Строка подключения JDBC для стандартного драйвера имеет следующую структуру:

jdbc:postgresql://<master_ip>[:<port>]/<database_name>

где:

  • <master_ip> — IP-адрес master-хоста ADB.

  • <port> — номер порта для подключения (5432 по умолчанию).

  • <database_name> — название БД в ADB для подключения.

Дополнительно ADB Spark Connector поддерживает использование сторонних драйверов JDBC. Чтобы использовать сторонний драйвер, выполните следующее:

  1. Подготовьте строку для подключения ODBC к master-хосту ADB.

  2. Загрузите JAR-файл со сторонним драйвером JDBC одним из следующих способов:

    • Используйте переменную окружения --jars при вызове spark-shell или spark-submit.

    • Выполните билд файла uber-jar со всеми зависимостями.

    • Установите JAR-файл со сторонним драйвером на всех нодах Spark.

  3. Укажите полный путь к Java-классу драйвера, используя опцию коннектора spark.adb.driver.

Использование коннектора

Для передачи данных между Spark и ADB с помощью ADB Spark Connector можно использовать стандартные утилиты spark-shell и spark-submit. Примеры использования spark3-shell приведены в статье Примеры использования ADB Spark Connector.

При разработке standalone-приложения Spark необходимо предварительно указать JAR-файл коннектора (/usr/lib/spark<version>/jars/adb-spark-connector-assembly-release-<connector_version>-spark-<spark_version>.jar) наряду с другими зависимостями приложения в файле uber-jar. Дополнительную информацию можно получить в статье Submitting Applications документации Spark.

Нашли ошибку? Выделите текст и нажмите Ctrl+Enter чтобы сообщить о ней