Работа с Yandex Object Storage из PySpark-задания с использованием кластера Apache Hive™ Metastore

Создайте сервисный аккаунт spark-agent для кластера Yandex Managed Service for Apache Spark™ с ролью managed-spark.integrationProvider — чтобы кластер Yandex Managed Service for Apache Spark™ мог взаимодействовать с другими ресурсами.

Создайте сервисный аккаунт metastore-agent с ролями managed-metastore.integrationProvider и storage.uploader — чтобы кластер Apache Hive™ Metastore мог взаимодействовать с другими ресурсами и экспортировать метаданные в бакет Object Storage.

Создайте бакеты:

• бакет для исходного кода PySpark-задания;
• бакет для выходных данных.

Предоставьте разрешения для сервисного аккаунта spark-agent на созданные бакеты:

• бакет для исходного кода PySpark-задания — разрешение READ;
• бакет для выходных данных — разрешение READ и WRITE.

Предоставьте разрешение READ и WRITE для сервисного аккаунта metastore-agent на бакет для выходных данных.

Создайте облачную сеть с именем integration-network.

Вместе с ней будут автоматически созданы три подсети в разных зонах доступности.

Для кластера Yandex Managed Service for Apache Spark™ создайте группу безопасности spark-sg в сети integration-network. Добавьте в группу следующее правило:

• Для исходящего трафика, чтобы разрешить подключение кластера Yandex Managed Service for Apache Spark™ к Apache Hive™ Metastore:

  • Диапазон портов — 9083.
  • Протокол — Любой (Any).
  • Назначение — CIDR.
  • CIDR блоки — 0.0.0.0/0.

Для кластера Apache Hive™ Metastore создайте группу безопасности metastore-sg в сети integration-network. Добавьте в группу следующие правила:

• Для входящего трафика от клиентов:

  • Диапазон портов — 30000-32767.
  • Протокол — Любой (Any).
  • Источник — CIDR.
  • CIDR блоки — 0.0.0.0/0.

• Для входящего трафика от балансировщика:

  • Диапазон портов — 10256.
  • Протокол — Любой (Any).
  • Источник — Проверки состояния балансировщика.

Создайте кластер Apache Hive™ Metastore с параметрами:

• Сервисный аккаунт — metastore-agent.
• Версия — 3.1.
• Сеть — integration-network.
• Подсеть — integration-network-kz1-a.
• Группы безопасности — metastore-sg.

Создайте кластер Yandex Managed Service for Apache Spark™ с параметрами:

• Сервисный аккаунт — spark-agent.
• Сеть — integration-network.
• Подсеть — integration-network-kz1-a.
• Группы безопасности — spark-sg.
• Metastore-сервер — созданный ранее кластер Apache Hive™ Metastore.

Создайте локально файл с именем job-create-table.py и скопируйте в него скрипт:

import random
import sys
from pyspark.sql import SparkSession


def prepare_table(spark, database, table):
    create_database_sql = "create database if not exists {database}"
    create_table_sql = """
    create table if not exists {database}.{table} (
        id int,
        value double
    )
    """
    truncate_table_sql = "truncate table {database}.{table}"

    spark.sql(create_database_sql.format(database=database))
    spark.sql(create_table_sql.format(database=database, table=table))
    spark.sql(truncate_table_sql.format(database=database, table=table))


def write_data(spark, database, table):
    data = [(i, random.random()) for i in range(100_000)]
    # Создание датафрейма
    df = spark.createDataFrame(data, schema=['id', 'value'])
    table_full_name = "{database}.{table}".format(database=database, table=table)
    df.write.mode('overwrite').format('json').saveAsTable(table_full_name)


def main():
    # Создание Spark-сессии
    spark = (
        SparkSession
        .builder
        .appName('job-create-table')
        .enableHiveSupport()
        .config('spark.sql.warehouse.dir', sys.argv[1])
        .getOrCreate()
    )
    database, table = 'database_1', 'table_1'
    prepare_table(spark, database, table)
    write_data(spark, database, table)


if __name__ == '__main__':
    if len(sys.argv) != 2:
        print("Usage: job-create-table s3a://<bucket>/<folder>", file=sys.stderr)
        sys.exit(-1)
    main()

В бакете для исходного кода создайте папку scripts и загрузите в нее файл job-create-table.py.

В бакете для выходных данных создайте папку warehouse, в которую будут загружены данные из БД database_1.

Создайте задание с параметрами:

• Тип задания — PySpark.
• Main python файл — s3a://<бакет_для_исходного_кода>/scripts/job-create-table.py.
• Аргументы — s3a://<бакет_для_выходных_данных>/warehouse.

Перейдите на страницу каталога и выберите сервис Managed Service for Apache Spark.

Нажмите на имя нужного кластера и выберите вкладку Задания.

Дождитесь, когда созданное PySpark-задание перейдет в статус Done.

Убедитесь, что в бакете для выходных данных, в папке warehouse, появился файл с данными из БД database_1.

Проверьте, что в кластере Apache Hive™ Metastore появились метаданные о БД database_1:

1. Экспортируйте метаданные из кластера Apache Hive™ Metastore в бакет для выходных данных.
2. Скачайте файл с метаданными и убедитесь, что в нем упоминается БД database_1.