Связаться с экспертомПопробовать бесплатно

Использование Yandex Object Storage в сервисе Yandex Managed Service for Apache Spark™

Статья создана
Обновлена 26 августа 2026 г.
Открыть в Markdown

Примечание

В регионе Казахстан доступна только зона доступности kz1-a.

В бакетах Yandex Object Storage можно хранить как файлы, необходимые для выполнения заданий в кластере Yandex Managed Service for Apache Spark™, так и результаты выполнения заданий.

Кластеры Yandex Managed Service for Apache Spark™ уже сконфигурированы для работы с S3-хранилищами. Чтобы использовать бакет Object Storage в PySpark-задании:

  • Предоставьте сервисному аккаунту кластера Yandex Managed Service for Apache Spark™ разрешение на доступ к бакету Object Storage.
  • Укажите путь к каталогу в бакете Object Storage в аргументах задания.

В этом руководстве показан пример работы с таблицей в бакете Object Storage из PySpark-задания с использованием встроенного локального каталога Hive.

Чтобы реализовать описанный пример:

  1. Подготовьте инфраструктуру.
  2. Подготовьте PySpark-задание.
  3. Проверьте результат.

Если созданные ресурсы вам больше не нужны, удалите их.

Примечание

Локальный каталог Hive позволяет обращаться к таблицам по имени без указания пути и использовать базовые возможности Hive, не подключая кластер Apache Hive™ Metastore. Хранящиеся в локальном каталоге метаданные остаются недоступными другим кластерам. Пример использования глобального каталога приведен в руководстве Работа с Yandex Object Storage из PySpark-задания с использованием кластера Apache Hive™ Metastore.

  • Бакеты Object Storage: использование хранилища и выполнение операций с данными (тарифы Object Storage).
  • Сервис Yandex Cloud Logging: объем записываемых данных и время их хранения (тарифы Cloud Logging).

Подготовьте инфраструктуру

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

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

    • <бакет_для_исходного_кода_PySpark_задания>.
    • <бакет_для_выходных_данных_PySpark_задания>.

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

    • <бакет_для_исходного_кода_PySpark_задания> — разрешение READ.
    • <бакет_для_выходных_данных_PySpark_задания> — разрешение READ и WRITE.

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

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

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

    • Сервисный аккаунт — spark-agent.
    • Сеть — spark-network.
    • Подсеть — spark-network-kz1-a.

Подготовьте PySpark-задание

Для PySpark-задания используется Python-скрипт, который создает таблицу table_1 в БД database_1. В аргументах PySpark-задания указан путь к каталогу, в котором будет создана БД. Для подключения встроенного каталога Hive в скрипте указывается параметр конфигурации Spark-сессии spark.sql.catalogImplementation=hive. Скрипт будет храниться в бакете Object Storage.

Подготовьте файл скрипта:

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

    job_save_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_save_table')
        .config('spark.executor.instances', 1)
        .config('spark.sql.warehouse.dir', sys.argv[1])
        .config('spark.sql.catalogImplementation', 'hive')
        .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-save-table s3a://<bucket>/<folder>", file=sys.stderr)
        sys.exit(-1)
    main()

  2. Создайте в бакете <бакет_для_исходного_кода_PySpark_задания> папку scripts и загрузите в нее файл job_save_table.py.

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

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

Проверьте результат

  1. В консоли управления выберите каталог.
  2. Перейдите в сервис Managed Service for Apache Spark.
  3. Нажмите на имя нужного кластера и выберите Задания.
  4. Дождитесь, когда созданное PySpark-задание перейдет в статус Done.
  5. Убедитесь, что в бакете <бакет_для_выходных_данных_PySpark_задания> в папке warehouse появилась БД database_1. Теперь данные из созданной БД хранятся в бакете Object Storage в формате JSON.

Удалите созданные ресурсы

Некоторые ресурсы платные. Удалите ресурсы, которые вы больше не будете использовать, чтобы не платить за них:

Предыдущая
Запуск PySpark-задания с помощью Yandex Managed Service for Apache Airflow™
Следующая
Интеграция Yandex Managed Service for Apache Spark™ и DataSphere