Импорт данных из Hive, S3, MongoDB и других систем

Импорт данных из внешних систем в YTsaurus реализован через SPYT.

На данной странице приведена инструкция по импорту данных с помощью скрипта import.py. Этот скрипт позволяет импортировать данные из Hive, Hadoop, S3, а также из СУБД, поддерживающих протокол JDBC. Для импорта из других систем хранения данных — например, из MongoDB — вы можете использовать SPYT напрямую, обратившись к этой системе с помощью соответствующего Spark Datasource.

Настройка подключения через Kerberos

Для работы с платформой Hadoop может использоваться протокол сетевой аутентификации Kerberos. Для настройки взаимодействия необходимо:

1. Создать на стороне Hadoop токен следующей командой:

$ fetchdt --renewer null /tmp/token

2. Скопировать токен в Кипарис:

$ cat /tmp/token | yt write-file //home/spark/conf/token

3. Задать при запуске приложения путь к токену в Кипарисе с помощью опции --files

Клиентский режим

При запуске в режиме --deploy-mode client для драйвера необходимо указать путь к локальному
файлу с токеном, так как процесс драйвера запускается не в кластере YTsaurus.
Для экзекьюторов указывается параметр `HADOOP_TOKEN_FILE_LOCATION=token````bash

$ spark-submit \
  --master ytsaurus:// \
  --deploy-mode client \
  --files yt://home/spark/conf/token \
  --conf spark.executorEnv.HADOOP_TOKEN_FILE_LOCATION=/tmp/token \
 ./script.py

Кластерный режим

Для запуска в режиме --deploy-mode cluster путь к файлу с токеном указывается через конфигурационный параметр spark.ytsaurus.config.global.path.

Дополнительно для драйвера необходимо добавить параметры kdc и realm. Пример команды:

$ spark-submit \
  --master ytsaurus:// \
  --deploy-mode cluster \
  --files yt://home/spark/conf/token \
  --conf spark.ytsaurus.config.global.path=//home/spark/conf/global \
  --conf spark.driver.extraJavaOption="-Djava.security.krb5.kdc=... -Djava.secutirykrb5.realm=..."  \
 ./script.py

Установите зависимости

jar-зависимости для чтения из Hive предоставляются вместе с пакетом pyspark. Пакеты для работы с SPYT (включая pyspark), должны быть установлены на системе, из которой вызывается импорт.

Для чтения из СУБД, которая поддерживает JDBC-протокол, нужно скачать JDBC-драйвер этой СУБД.

Файл connectors/pom.xml содержит конфиг для Maven, в котором указаны JDBC-драйвера для MySQL и PostgreSQL. Эти драйвера можно скачать следующим образом:

~/yt/connectors$ mvn dependency:copy-dependencies

mvn скачает jar-файлы с драйверами для PostgreSQL и MySQL в директорию target/dependency.

Если необходимо импортировать данные из другой СУБД, добавьте JDBC-драйвер этой СУБД в качестве зависимости в pom.xml, и запустите команду $ mvn dependency:copy-dependencies

Запустите импорт данных

Для импорта данных следует запустить скрипт import.py:

$ ./import.py \
   # Опции импорта.
   # ...

Опции должны идентифицировать:

• источник данных;
• путь к импортируемым данным;
• путь в YTsaurus, куда должны быть записаны импортируемые данные.

Список всех опций приведён в конце этого раздела, в секции Опции. Ниже приведены примеры импорта данных из разных систем.

Клиентский и кластерный режим

import.py можно запускать в двух режимах:

• Клиентский (по умолчанию) — python import.py ...: драйвер Spark стартует локально на машине, откуда запущен скрипт, а экзекьюторы — в кластере YTsaurus.

• Кластерный — import.py запускается как Spark-приложение через spark-submit ... --deploy-mode cluster, при этом драйвер тоже стартует внутри кластера YTsaurus. Режим определяется автоматически (по факту запуска через spark-submit), отдельный флаг не нужен:

  $ spark-submit \
      --master ytsaurus://<proxy> \
      --deploy-mode cluster \
      --jars yt:///path/to/jars/driver.jar \
      import.py \
      --jdbc postgresql --jdbc-server pg_host:5432 --jdbc-user user --jdbc-password '<password>' \
      --input jdbc:database_name.table_name \
      --output //path/in/yt/table
  

Кластерный режим удобен при сабмите с внешней машины (например, вне Kubernetes-кластера, в котором развёрнут YTsaurus): драйвер уходит внутрь кластера, и не требуется сетевая доступность драйвер↔экзекьюторы с внешней машины.

Сетевое взаимодействие внутри Kubernetes

Если сабмит идёт с внешней машины, а YTsaurus развёрнут в Kubernetes, снаружи обычно доступен только внешний прокси, а внутрикластерный трафик (драйвер и экзекьюторы ↔ YTsaurus) должен идти через внутренний прокси. Чтобы это обеспечить, задайте параметр spark.hadoop.yt.clusterProxy — адрес прокси, доступного изнутри кластера; при этом --master указывает на внешний адрес, по которому выполняется сабмит:

$ spark-submit \
    --master ytsaurus://<внешний proxy> \
    --deploy-mode cluster \
    --conf spark.hadoop.yt.clusterProxy=<внутренний proxy в k8s> \
    import.py ...

В клиентском режиме тот же параметр передаётся через --extra-conf spark.hadoop.yt.clusterProxy=....

Hive

Чтобы импортировать данные из Hive:

1. Загрузите на Кипарис файлы hadoop-aws-*.jar и aws-java-sdk-bundle-*.jar из локальной директории target/dependency.
2. Запустите импорт данных следующей командой:

   $ ./import.py \
     --metastore master_host:9083 \
     --warehouse-dir /path/to/hive/warehouse \
     --input hive:database_name.table_name \
     --output //path/in/yt/table \
     --proxy https://ytsaurus.company.net \
     --num-executors 5 \
     --executor-memory 4G \
     --executor-memory-overhead 2G \
     --jars yt:///path/to/jars/aws-java-sdk-bundle-1.11.901.jar \
            yt:///path/to/jars/hadoop-aws-3.3.1.jar
   
   Перейти к описанию опций.

Альтернативно, можно передать в Hive SQL-запрос для исполнения и импортировать результат этого запроса. Для этого используйте префикс hive_sql для описания исходных данных:

$ ./import.py \
    ...
    --input hive_sql:database_name:SELECT * FROM action_log WHERE action_date > '2023-01-01' \
    ...

HDFS

Для импорта файла из HDFS используйте спецификатор с указанием формата этого файла, а также адрес HDFS NameNode:

$ ./import.py \
    ...
    --input text:hdfs://namenode/path/to/text/file
    ...

import.py также поддерживает форматы parquet, orc.

СУБД, поддерживающие JDBC-протокол

Чтобы импортировать данные из JDBC-совместимых СУБД (например, из PostgreSQL), запустите команду:

$ ./import.py \
    --jdbc postgresql \
    --jdbc-server pg_host:5432 \
    --jdbc-user user \
    --jdbc-password '' \  # Получить пароль из консольного ввода
    --input jdbc:database_name.table_name \
    --output //path/in/yt/table

Перейти к описанию опций.

Чтобы передать в СУБД запрос для исполнения и импортировать результат, можно использовать префикс jdbc_sql:

$ ./import.py \
    ...
    --input jdbc_sql:database_name:SELECT * FROM users WHERE signup_date > '2023-01-01' \
    ...

Параллельное чтение больших таблиц

По умолчанию import.py читает JDBC-источник в один поток (одним запросом). Для больших таблиц это медленно и может приводить к ошибкам долгих запросов на стороне СУБД (например, ORA-01555: snapshot too old в Oracle). Чтобы читать таблицу параллельно, задайте числовую или дата-колонку разбиения и её диапазон:

$ ./import.py \
    --jdbc postgresql \
    --jdbc-server pg_host:5432 \
    --jdbc-user user \
    --jdbc-password '' \
    --input jdbc:database_name.table_name \
    --jdbc-partition-column id \
    --jdbc-lower-bound 1 \
    --jdbc-upper-bound 1000000 \
    --jdbc-num-partitions 16 \
    --output //path/in/yt/table

SPYT разобьёт чтение на --jdbc-num-partitions диапазонов вида column >= a AND column < b и прочитает их параллельно. Границы --jdbc-lower-bound/--jdbc-upper-bound задают только шаг разбиения — данные они не фильтруют (значения за пределами попадают в крайние партиции). Тип колонки должен быть числовым или датой; желательно проиндексированной или являющейся ключом секционирования таблицы, чтобы каждый запрос читал только свою часть.

Опции партиционирования глобальные — они применяются одинаково ко всем таблицам, перечисленным через --input. Если за один запуск импортируется несколько таблиц, колонка разбиения и границы должны подходить каждой из них; иначе запускайте import.py отдельно под каждую таблицу.

S3

Чтобы импортировать данные из S3:

1. Загрузите на Кипарис файлы hadoop-aws-*.jar и aws-java-sdk-bundle-*.jar из локальной директории target/dependency.

2. Запустите импорт данных следующей командой:

   ./import.py \
    --input parquet:s3a://bucket/path/to/data/sample-parquet \
    --output //home/tables/import_from_s3 \
    --s3-access-key <S3 Access Key> \
    --s3-secret-key <S3 Secret Key> \
    --s3-endpoint <S3 endpoint> \
    --proxy https://ytsaurus.company.net \
    --num-executors 5 \
    --executor-memory 4G \
    --executor-memory-overhead 2G \
    --jars yt:///path/to/jars/aws-java-sdk-bundle-1.11.901.jar \
         yt:///path/to/jars/hadoop-aws-3.3.1.jar
   

   Описание опций приведено ниже.

Опции

import.py поддерживает следующие опции:

Если операция импорта данных предусматривает запуск кластера SPYT, следующие опции командной строки могут использоваться для конфигурации этого кластера:

В случае импорта данных с S3:

Поддерживаются следующие спецификаторы для импортируемых данных:

По умолчанию при записи таблицы в YTsaurus подразумевается, что таблица не существует. Если таблица уже существует, можно её перезаписать или дополнить, используя спецификатор overwrite или append. Например: --output overwrite:/path/to/yt

Преобразования типов

Импорт сложных типов из внешних систем может поддерживаться лишь частично. Система типов YTsaurus не имеет взаимно однозначного соответствия с системами типов других систем хранения данных. При импорте SPYT сохранит тип по мере возможности, и преобразует значение в строку в тех случаях, когда не удалось вывести правильный тип. При необходимости используйте SQL для корректного преобразования типов.

Диапазоны значений одного типа могут оказаться разными. Например, YTsaurus date не хранит даты ранее Юникс-эпохи, 1 января 1970 года. Попытка записать в YTsaurus более ранние даты приведёт к ошибке. Даты раньше Юникс-эпохи в YTsaurus можно хранить как строки (например, используя преобразование to_char(date_value, 'YYYY-MM-DD') в PostgreSQL), или как целые числа (date_value - '1970-01-01' в PostgreSQL).