Подключение из Apache Spark#

Процесс подключения к Iceberg REST Catalog из Apache Spark описан в официальной документации Iceberg.

Данный документ разъясняет особенности конфигурации Apache Spark для подключения к CedrusData Catalog.

Подготовка#

  1. Убедитесь, что CedrusData Catalog содержит хотя бы один Iceberg каталог

  2. Подключите к Spark актуальную версию библиотеки iceberg-spark-runtime согласно официальной документации. При подключении библиотеки необходимо убедиться, Spark и Iceberg используют одну и туже версию Scala.

  3. Получите постоянный или временный токен доступа CedrusData Catalog

Конфигурация Apache Spark#

Для подключения к CedrusData Catalog необходимо задать параметры сессии, описанные ниже.

spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.<имя_каталога_spark>=org.apache.iceberg.spark.SparkSessionCatalog
spark.sql.catalog.<имя_каталога_spark>.type=rest
spark.sql.catalog.<имя_каталога_spark>.uri=<адрес_CedrusData_Catalog>
spark.sql.catalog.<имя_каталога_spark>.warehouse=<имя_каталога_Iceberg_в_CedrusData_Catalog>
spark.sql.catalog.<имя_каталога_spark>.<тип_аутентификации>=<токен_доступа_CedrusData_Catalog>

Где:

  • <имя_каталога_spark> — произвольное имя каталога Apache Spark, который вы будете использовать при выполнении SQL-запросов. Не имеет отношения к именам реальных объектов Apache Iceberg.

  • <адрес_CedrusData_Сatalog> — адрес, по которому доступен Iceberg REST Catalog интерфейс CedrusData Catalog (http(s)://host[:port]/catalog/iceberg)

  • <имя_каталога_Iceberg_в_CedrusData_Catalog> — имя объекта Iceberg catalog, зарегистрированного в CedrusData Catalog

  • <тип_аутентификации> — механизм аутентификации в CedrusData Catalog (credential или token)

  • <токен_доступа_CedrusData_Catalog> — токен доступа CedrusData Catalog: постоянный токен доступа для типа аутентификации credential; постоянный иди временный токен доступа для типа аутентификации token

Кроме этого, вы должны задать конфигурацию FileIO для доступа к файловой системе. В конце документа приведен пример конфигурации FileIO для доступа к S3-совместимой файловой системе.

Пример#

Инструкции ниже описывают, как интегрировать локальный экземпляр Apache Spark с CedrusData Catalog.

  1. Создайте файловую систему CedrusData Catalog на основе MinIO согласно инструкциям. Не останавливайте Docker-контейнеры.

  2. Создайте Iceberg каталог с именем ice_prod в CedrusData Catalog:

    docker exec -it example-catalog catalog iceberg catalog create \
  --catalog-name ice_prod \
  --file-system-name minio \
  --file-system-location s3://ice-bucket

  3. Скачайте дополнительные библиотеки, которые потребуются Apache Spark для взаимодействия с CedrusData Catalog и MinIO. В обычных условиях вы будете делать это с помощью опции Apache Spark --packages. В данном примере мы запускаем Apache Spark из официального Docker-контейнера, в котором опция --packages может не работать в ряде случаев. Поэтому мы предпочитаем скачать все необходимые библиотеки на хост, и потом передать их в Docker-контейнер Spark через mount:

    mkdir jars
wget https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.7.0/iceberg-spark-runtime-3.5_2.12-1.7.0.jar -O jars/iceberg-spark-runtime-3.5_2.12-1.7.0.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/s3/2.29.17/s3-2.29.17.jar -O jars/s3-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/aws-core/2.29.17/aws-core-2.29.17.jar -O jars/aws-core-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/sdk-core/2.29.17/sdk-core-2.29.17.jar -O jars/sdk-core-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/auth/2.29.17/auth-2.29.17.jar -O jars/auth-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/identity-spi/2.29.17/identity-spi-2.29.17.jar -O jars/identity-spi-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/utils/2.29.17/utils-2.29.17.jar -O jars/utils-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/http-client-spi/2.29.17/http-client-spi-2.29.17.jar -O jars/http-client-spi-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/sts/2.29.17/sts-2.29.17.jar -O jars/sts-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/glue/2.29.17/glue-2.29.17.jar -O jars/glue-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/kms/2.29.17/kms-2.29.17.jar -O jars/kms-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/dynamodb/2.29.17/dynamodb-2.29.17.jar -O jars/dynamodb-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/regions/2.29.17/regions-2.29.17.jar -O jars/regions-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/apache-client/2.29.17/apache-client-2.29.17.jar -O jars/apache-client-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/retries-spi/2.29.17/retries-spi-2.29.17.jar -O jars/retries-spi-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/profiles/2.29.17/profiles-2.29.17.jar -O jars/profiles-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/endpoints-spi/2.29.17/endpoints-spi-2.29.17.jar -O jars/endpoints-spi-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/http-auth-spi/2.29.17/http-auth-spi-2.29.17.jar -O jars/http-auth-spi-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/http-auth-aws/2.29.17/http-auth-aws-2.29.17.jar -O jars/http-auth-aws-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/http-auth/2.29.17/http-auth-2.29.17.jar -O jars/http-auth-2.29.17.jar && \
wget https://repo1.maven.org/maven2/org/reactivestreams/reactive-streams/1.0.4/reactive-streams-1.0.4.jar -O jars/reactive-streams-1.0.4.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/retries/2.29.17/retries-2.29.17.jar -O jars/retries-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/metrics-spi/2.29.17/metrics-spi-2.29.17.jar -O jars/metrics-spi-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/aws-xml-protocol/2.29.17/aws-xml-protocol-2.29.17.jar -O jars/aws-xml-protocol-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/aws-query-protocol/2.29.17/aws-query-protocol-2.29.17.jar -O jars/aws-query-protocol-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/protocol-core/2.29.17/protocol-core-2.29.17.jar -O jars/protocol-core-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/checksums/2.29.17/checksums-2.29.17.jar -O jars/checksums-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/checksums-spi/2.29.17/checksums-spi-2.29.17.jar -O jars/checksums-spi-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/json-utils/2.29.17/json-utils-2.29.17.jar -O jars/json-utils-2.29.17.jar && \
wget https://repo1.maven.org/maven2/software/amazon/awssdk/third-party-jackson-core/2.29.17/third-party-jackson-core-2.29.17.jar -O jars/third-party-jackson-core-2.29.17.jar

  4. Запустите SQL-консоль Apache Spark в Docker-контейнере. Наш каталог Spark для работы с Iceberg имеет имя ice:

    CATALOG_ACCESS_TOKEN=$(docker exec example-catalog cat /data/catalog/admin-access-token) && \
docker run -it --rm --name example-spark --network=example-network \
  -v $PWD/jars/iceberg-spark-runtime-3.5_2.12-1.7.0.jar:/opt/spark/jars/iceberg-spark-runtime-3.5_2.12-1.7.0.jar \
  -v $PWD/jars/s3-2.29.17.jar:/opt/spark/jars/s3-2.29.17.jar \
  -v $PWD/jars/aws-core-2.29.17.jar:/opt/spark/jars/aws-core-2.29.17.jar \
  -v $PWD/jars/sdk-core-2.29.17.jar:/opt/spark/jars/sdk-core-2.29.17.jar \
  -v $PWD/jars/auth-2.29.17.jar:/opt/spark/jars/auth-2.29.17.jar \
  -v $PWD/jars/identity-spi-2.29.17.jar:/opt/spark/jars/identity-spi-2.29.17.jar \
  -v $PWD/jars/utils-2.29.17.jar:/opt/spark/jars/utils-2.29.17.jar \
  -v $PWD/jars/http-client-spi-2.29.17.jar:/opt/spark/jars/http-client-spi-2.29.17.jar \
  -v $PWD/jars/sts-2.29.17.jar:/opt/spark/jars/sts-2.29.17.jar \
  -v $PWD/jars/glue-2.29.17.jar:/opt/spark/jars/glue-2.29.17.jar \
  -v $PWD/jars/kms-2.29.17.jar:/opt/spark/jars/kms-2.29.17.jar \
  -v $PWD/jars/dynamodb-2.29.17.jar:/opt/spark/jars/dynamodb-2.29.17.jar \
  -v $PWD/jars/regions-2.29.17.jar:/opt/spark/jars/regions-2.29.17.jar \
  -v $PWD/jars/apache-client-2.29.17.jar:/opt/spark/jars/apache-client-2.29.17.jar \
  -v $PWD/jars/retries-spi-2.29.17.jar:/opt/spark/jars/retries-spi-2.29.17.jar \
  -v $PWD/jars/profiles-2.29.17.jar:/opt/spark/jars/profiles-2.29.17.jar \
  -v $PWD/jars/endpoints-spi-2.29.17.jar:/opt/spark/jars/endpoints-spi-2.29.17.jar \
  -v $PWD/jars/http-auth-spi-2.29.17.jar:/opt/spark/jars/http-auth-spi-2.29.17.jar \
  -v $PWD/jars/http-auth-aws-2.29.17.jar:/opt/spark/jars/http-auth-aws-2.29.17.jar \
  -v $PWD/jars/http-auth-2.29.17.jar:/opt/spark/jars/http-auth-2.29.17.jar \
  -v $PWD/jars/reactive-streams-1.0.4.jar:/opt/spark/jars/reactive-streams-1.0.4.jar \
  -v $PWD/jars/retries-2.29.17.jar:/opt/spark/jars/retries-2.29.17.jar \
  -v $PWD/jars/metrics-spi-2.29.17.jar:/opt/spark/jars/metrics-spi-2.29.17.jar \
  -v $PWD/jars/aws-xml-protocol-2.29.17.jar:/opt/spark/jars/aws-xml-protocol-2.29.17.jar \
  -v $PWD/jars/aws-query-protocol-2.29.17.jar:/opt/spark/jars/aws-query-protocol-2.29.17.jar \
  -v $PWD/jars/protocol-core-2.29.17.jar:/opt/spark/jars/protocol-core-2.29.17.jar \
  -v $PWD/jars/checksums-2.29.17.jar:/opt/spark/jars/checksums-2.29.17.jar \
  -v $PWD/jars/checksums-spi-2.29.17.jar:/opt/spark/jars/checksums-spi-2.29.17.jar \
  -v $PWD/jars/json-utils-2.29.17.jar:/opt/spark/jars/json-utils-2.29.17.jar \
  -v $PWD/jars/third-party-jackson-core-2.29.17.jar:/opt/spark/jars/third-party-jackson-core-2.29.17.jar \
  spark /opt/spark/bin/spark-sql \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.ice=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.ice.type=rest \
  --conf spark.sql.catalog.ice.uri=http://example-catalog:9080/catalog/iceberg \
  --conf spark.sql.catalog.ice.warehouse=ice_prod \
  --conf spark.sql.catalog.ice.token=$CATALOG_ACCESS_TOKEN \
  --conf spark.sql.catalog.ice.io-impl=org.apache.iceberg.aws.s3.S3FileIO \
  --conf spark.sql.catalog.ice.s3.endpoint=http://example-minio:9000 \
  --conf spark.sql.catalog.ice.s3.access-key-id=accesskey \
  --conf spark.sql.catalog.ice.s3.secret-access-key=secretkey \
  --conf spark.sql.catalog.ice.s3.path-style-access=true \
  --conf spark.sql.catalog.ice.client.region=us-east-1

  5. Выполните ряд команд SQL:

    CREATE SCHEMA IF NOT EXISTS ice.hr;
DROP TABLE IF EXISTS ice.hr.employee;
CREATE TABLE ice.hr.employee(id BIGINT, name VARCHAR(100));
INSERT INTO ice.hr.employee VALUES (1, 'John');
SELECT * FROM ice.hr.employee;

  6. Пройдите в консоль MinIO http://localhost:9001 (имя пользователя accesskey, пароль secretkey), и ознакомьтесь с созданными файлам Iceberg в директории ice-bucket/hr/employee.

  7. Получите список таблиц Iceberg через утилиту командной строки CedrusData Catalog:

    docker exec -it example-catalog catalog iceberg table list

  8. (при необходимости) Остановите контейнер Spark с помощью команды exit; или комбинации клавиш Ctrl+C. Удалите директорию jars/. Очистите остальные ресурсы, следуя последнему шагу инструкции.