Подключение DBT к Managed Spark через Spark Connect

Перед началом работы

1. Зарегистрируйтесь в личном кабинете Cloud.ru.

   Если вы уже зарегистрированы, войдите под своей учетной записью.

2. Настройте DNS-сервер и подсеть.

3. Создайте кластер Data Platform, в котором будут размещены инстансы.

4. Создайте бакет Object Storage с названием dbt-labs.

   1. В бакете создайте папку raw.

   2. Скачайте файл с данными orders.csv и загрузите его в папку raw.

5. Создайте инстанс Managed Metastore. Он необходим для управления метаданными о таблицах, загруженных в Object Storage.

   1. В блоке Источник выберите Object Storage.

   2. В поле Путь до warehouse укажите бакет dbt-labs.

6. Создайте инстанс Managed Spark. В блоке Настройки доступа активируйте опцию Подключить публичный хост.

7. Дождитесь, когда статус инстанса Managed Spark сменится на «Готов».

8. Создайте Spark Connect. Активируйте опцию Добавить Spark конфигурацию (–conf) и укажите следующие параметры и их значения:

   Аргумент	Значение
   spark.sql.catalogImplementation	hive
   spark.sql.warehouse.dir	s3a://dbt-labs/
   spark.hadoop.hive.metastore.uris	Thrift URL из карточки созданного ранее инстанса Managed Metastore

9. Создайте API-ключ. В поле Сервисы выберите Evolution Data Platform. Сохраните значение API-ключа, после закрытия окна получить его будет нельзя.

10. Создайте секрет. В поле Значение секрета выберите Стандартный режим и введите сохраненное ранее значение API-ключа.

11. Создайте ноутбук на основе базового образа VSCode-Server.

    1. В блоке Окружение и данные добавьте следующие переменные и их значения:

       Ключ	Значение
       host	Публичный хост из карточки созданного ранее инстанса Managed Spark
       port	443

    2. В блоке Окружение и данные добавьте переменную с секретом.

       1. В поле Ключ введите apikey.

       2. В поле Секрет выберите созданный ранее секрет.

1. Настройте окружение

1. Откройте терминал: нажмите и выберите View → Terminal.

2. Создайте рабочую папку dbt:

   mkdir dbt
   
   

3. В папке dbt создайте файл requirements.txt со следующим содержимым:

   dbt-spark==1.9.3
   pyspark[connect]==3.5.4
   pandas>=1.0.5
   pyarrow>=4.0.0
   grpcio>=1.48.1
   ipykernel>=7.2.0
   
   

4. Создайте виртуальное окружение:

   python -m venv .venv
   
   

5. Активируйте виртуальное окружение:

   source .venv/bin/activate
   
   

6. Установите зависимости:

   pip install -r  requirements.txt
   
   

7. Скачайте корневой сертификат:

   mkdir ./certs && wget -O ./certs/dp_cert.crt  https://xbox.cloud.ru/index.php/s/wEso6nB7bKYxkkf/download  --no-check-certificate
   
   

8. Добавьте путь к сертификату в переменную окружения:

   export GRPC_DEFAULT_SSL_ROOTS_FILE_PATH=/home/jovyan/dbt/certs/dp_cert.crt
   
   

2. Создайте внешнюю таблицу

1. В папке dbt создайте файл spark_connect.ipynb.

2. В файле spark_connect.ipynb добавьте ячейку и вставьте в нее скрипт для создания Spark Connect:

   import os
   from pyspark.sql import SparkSession
   host = os.environ["host"]
   port = os.environ["port"]
   apikey = os.environ["apikey"]
   os.environ["GRPC_DEFAULT_SSL_ROOTS_FILE_PATH"] = "./certs/dp_cert.crt"
   spark = SparkSession.builder.remote(f"sc://{host}:{port}/;use_ssl=true;x-api-key={apikey}").getOrCreate()
   
   

3. Запустите код в ячейке.

4. Добавьте вторую ячейку и вставьте в нее скрипт для создания внешней таблицы:

   # Создайте базу данных в hive metastore
   spark.sql("CREATE DATABASE IF NOT EXISTS raw")
   # Создайте внешнюю таблицу, для чтения данных из ObjectStorage
   spark.sql("""
          CREATE EXTERNAL TABLE IF NOT EXISTS raw.orders (
             order_id INT,
             order_date DATE,
             customer_id STRING,
             product STRING,
             category STRING,
             quantity INT,
             unit_price DOUBLE
          )
          USING CSV
          OPTIONS (
             path 's3a://dbt-labs/raw/',
             delimiter ',',
             header 'true'
          )
   """)
   # Проверьте, что данные доступны
   spark.sql("SELECT * FROM raw.orders").show(truncate=False)
   
   

5. Запустите код во второй ячейке.

3. Подготовьте DBT-проект

1. В терминале выполните команду:

   dbt init dbt_spark_lab --profiles-dir .
   
   

2. В диалоге настройки укажите следующие значения параметров:

   1. Which database would you like to use? — spark.

   2. Host — locahost.

   3. Desired authentication method option — thrift.

   4. Port — 443.

   5. Schema (default schema that dbt will build objects in) — default.

   6. Threads — 1.

3. Откройте файл /dbt_spark_lab/profiles.yml. В строке method: thrift замените thrift на session.

4. Подготовьте папку /dbt_spark_lab/models/:

   1. Удалите папку example и файл schema.yml.

   2. Загрузите файл sources.yaml.

   3. Cоздайте папку /dbt_spark_lab/models/marts.

   4. Загрузите файл marts_sales_by_category.sql в папку /dbt_spark_lab/models/marts.

4. Запустите DBT

1. В терминале перейдите в папку /dbt_spark_lab :

   cd dbt_spark_lab
   
   

2. Создайте переменную окружения для подключения к Spark Connect:

   export SPARK_REMOTE="sc://${host}:${port}/;use_ssl=true;x-api-key=${apikey}"
   
   

3. Запустите DBT-проект:

   dbt build
   
   

5. Проверьте работу DBT

1. Перейдите в файл spark_connect.ipynb.

2. Добавьте ячейку и выполните запрос:

   spark.sql("SELECT * FROM marts_sales_by_category").show(truncate=False)
   
   

Результат