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

Работа с Yandex Object Storage из PySpark-задания без использования Hive

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

Важно

Часть ресурсов, необходимых для прохождения практического руководства, доступны только в регионе Россия.

В бакетах 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. Этот подход имеет следующие ограничения:

  • доступ к данным возможен только по путям в Object Storage, но не по именам таблиц;
  • метаданные о таблицах не создаются и не сохраняются.

Примечание

Для того чтобы обращаться к данным по имени таблицы и хранить метаданные таблиц, используйте локальный или глобальный каталог Hive:

  • Локальный каталог — для работы с данными и метаданными в пределах одного кластера.
  • Глобальный каталог (кластер Apache Hive™ Metastore) — для работы с общими метаданными таблиц из нескольких кластеров Yandex Managed Service for Apache Spark™.

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

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

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

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

  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.

Подготовьте файл с данными

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

    id,value
1,10
2,20
3,30
4,40
5,50

  2. Создайте в бакете <бакет_для_данных_PySpark_задания> папку input и загрузите в нее файл input.csv.

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

Для PySpark-задания используется Python-скрипт, который:

  • считывает CSV-файл из бакета Object Storage;
  • регистрирует DataFrame как временное представление Spark SQL;
  • выполняет SQL-запрос;
  • сохраняет результат в бакете Object Storage в формате Parquet.

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

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

    job_process_data.py
    import sys
from pyspark.sql import SparkSession


def main():
    if len(sys.argv) != 3:
        print(
            "Usage: job_process_data <input_path> <output_path>",
            file=sys.stderr
        )
        sys.exit(-1)

    input_path = sys.argv[1]
    output_path = sys.argv[2]

    spark = (
        SparkSession.builder
        .appName("object-storage-example")
        .config("spark.executor.instances", 1)
        .getOrCreate()
    )

    df = spark.read.load(
        input_path,
        format="csv",
        header=True,
        inferSchema=True
    )

    df.createOrReplaceTempView("source_data")

    result_df = spark.sql("""
        SELECT
            id,
            value
        FROM source_data
        WHERE value >= 30
    """)

    result_df.write.mode("overwrite").save(
        output_path,
        format="parquet"
    )

    spark.stop()


if __name__ == "__main__":
    main()

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

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

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

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

  1. В консоли управления выберите каталог.
  2. Перейдите в сервис Managed Service for Apache Spark.
  3. Нажмите на имя нужного кластера и выберите вкладку Задания.
  4. Дождитесь, когда созданное PySpark-задание перейдет в статус Done.
  5. Убедитесь, что в бакете <бакет_для_данных_PySpark_задания> в папке output появились файлы *.parquet. Результат обработки данных доступен другим Spark-заданиям по пути s3a://<бакет_для_данных_PySpark_задания>/output.

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

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

Предыдущая
Обзор
Следующая
Использование Yandex Object Storage в Yandex Managed Service for Apache Spark™