Совместная работа с таблицами Yandex Data Processing с использованием Metastore

Создайте сервисный аккаунт с именем dataproc-s3-sa и назначьте ему роли dataproc.agent и dataproc.provisioner.

В Object Storage создайте бакеты и настройте доступ к ним:

1. Создайте бакет для исходных данных и предоставьте сервисному аккаунту кластера разрешение READ для этого бакета.
2. Создайте бакет для результатов обработки и предоставьте сервисному аккаунту кластера разрешение READ и WRITE для этого бакета.

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

В сети dataproc-network создайте подсеть в любой зоне доступности.

Настройте NAT-шлюз для созданной подсети.

Создайте два кластера Yandex Data Processing с именами dataproc-source и dataproc-target, с любой подходящей конфигурацией хостов и следующими настройками:

• Сервисы:
  • SPARK;
  • YARN.
• Сервисный аккаунт — dataproc-sa.
• Свойства — spark:spark.sql.hive.metastore.sharedPrefixes со значением com.amazonaws,ru.yandex.cloud. Нужно для выполнения заданий PySpark и для интеграции с Metastore.
• Имя бакета — бакет, который вы создали для выходных данных.
• Сеть — dataproc-network.

Если в облачной сети используются группы безопасности, добавьте в группу безопасности кластеров Yandex Data Processing следующее правило для исходящего трафика:

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

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

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

   create-table.py

   from pyspark.sql.types import *
   from pyspark.sql import SparkSession
   
   # Создание Spark-сессии
   spark = SparkSession.builder \
       .appName("create-table") \
       .enableHiveSupport() \
       .getOrCreate()
   
   # Создание схемы данных
   schema = StructType([StructField('Name', StringType(), True),
   StructField('Capital', StringType(), True),
   StructField('Area', IntegerType(), True),
   StructField('Population', IntegerType(), True)])
   
   # Создание датафрейма
   df = spark.createDataFrame([('Австралия', 'Канберра', 7686850, 19731984), ('Австрия', 'Вена', 83855, 7700000)], schema)
   
   # Запись датафрейма в бакет в виде таблицы countries
   df.write.mode("overwrite").option("path","s3a://<имя_выходного_бакета>/countries").saveAsTable("countries")
   

2. Укажите в скрипте имя выходного бакета, в который будет сохранен файл с таблицей countries.

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

Создайте задание PySpark, указав в поле Main python файл путь к файлу скрипта: s3a://<имя_входного_бакета>/scripts/create-table.py.

Дождитесь завершения задания и проверьте, что в выходном бакете в папке countries появился файл part-00000-....

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

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

   obtain-table.py

   from pyspark.sql import SparkSession
   
   # Создание Spark-сессии
   spark = SparkSession.builder \
       .appName("obtain-table") \
       .enableHiveSupport() \
       .getOrCreate()
   
   spark.catalog.listDatabases()
   
   # Получение информации о таблице countries из Metastore
   df = spark.sql("describe extended countries")
   
   # Запрос данных из таблицы countries
   df = spark.sql("select * from countries")
   
   # Перенос таблицы в бакет для проверки
   df.repartition(1).write.csv(path='s3a://<имя_выходного_бакета>/csv', header=True, sep=',')
   

2. Укажите в скрипте имя выходного бакета, в который будет сохранен CSV-файл с таблицей countries.

3. Загрузите файл obtain-table.py в бакет для входных данных в папку scripts.

Создайте задание PySpark, указав в поле Main python файл путь к файлу скрипта: s3a://<имя_входного_бакета>/scripts/obtain-table.py.

Дождитесь выполнения задания и убедитесь, что в выходном бакете появилась папка csv с таблицей в формате CSV.