Работа с пользовательским образом

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

1. Зарегистрируйтесь в личном кабинете Cloud.ru. Если вы уже зарегистрированы, войдите под своей учетной записью.

2. Создайте бакет Object Storage, в котором будут храниться необходимые файлы и логи.

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

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

5. Скачайте и установите root-сертификат на устройство.

6. Создайте пароль и добавьте его в Secret Manager. Этот секрет станет паролем для доступа к интерфейсу Managed Spark.

7. Создайте инстанс Managed Spark.

8. Создайте реестр Artifact Registry, в котором будет храниться пользовательский образ Managed Spark.

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

1. Скачайте CSV-таблицы client-spark-image.csv и sales-spark-image.csv: нажмите Скачать в правом верхнем углу.

2. В ранее созданном бакете Object Storage создайте папку input.

3. Загрузите CSV-таблицы в папку input.

2. Подготовьте скрипт задачи

1. Скопируйте скрипт и назовите файл script-spark-image.py.

   import numpy as np
   import time
   
   from pyspark.sql import SparkSession
   from pyspark.sql.types import FloatType
   from pyspark.sql.functions import lit, udf
   
   bucket_name = 'your-bucket-name'
   
   spark = (SparkSession.builder
            .appName("sales")
            .getOrCreate()
            )
   
   # Read the source data from csv
   df_sales = spark.read \
   .format("csv") \
   .option("header", "true") \
   .option("inferSchema", "true") \
   .option("delimiter", ";") \
   .load(f"s3a://{bucket_name}/input/sales-spark-image.csv")
   
   df_client = spark.read \
   .format("csv") \
   .option("header", "true") \
   .option("inferSchema", "true") \
   .option("delimiter", ";") \
   .load(f"s3a://{bucket_name}/input/client-spark-image.csv")
   
   # get average cost for all sales
   np_arr = np.array(df_sales.select('sales').collect())
   avg = np.average(np_arr)
   print(f'Average cost: {avg}')
   # define UDF
   
   @udf(returnType=FloatType())
   def calc_diff_avg(avg, val):
         return val - avg
   # Create result with sale price and diff between sale price and average price
   
   df_result = df_sales \
   .join(df_client, df_sales.order_number ==  df_client.order_number,"inner") \
   .select( \
         df_client.order_number, \
         df_client.order_date, \
         df_client.phone, \
         df_client.address_line1, \
         df_client.address_line2, \
         df_client.city, \
         df_client.state, \
         df_client.postal_code, \
         df_client.country, \
         df_client.territory, \
         df_client.contact_last_name, \
         df_client.contact_first_name, \
         df_client.deal_size, \
         df_client.car, \
         df_sales.sales, \
         calc_diff_avg(lit(avg), df_sales.sales).alias("diff_with_avg") \
   )
   
   # Write the result to csv file
   df_result.write.mode('overwrite').option("header","true").csv(f"s3a://{bucket_name}/output/sales")
   
   

2. В строке bucket_name = 'your-bucket-name' замените your-bucket-name на название бакета Object Storage.

3. В ранее созданном бакете Object Storage создайте папку jobs.

4. Загрузите скрипт в папку jobs.

• <bucket>

  • input

    • sales-spark-image.csv

    • client-spark-image.csv

  • jobs

    • script-spark-image.py

3. Подготовьте образ в Artifact Registry

1. Создайте Dockerfile для сборки образа.

   FROM apache/spark:3.5.0-scala2.12-java11-python3-ubuntu
   
   # add S3 libs
   RUN curl https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.12.262/aws-java-sdk-bundle-1.12.262.jar -o /opt/spark/jars/aws-java-sdk-bundle-1.12.262.jar
   RUN curl https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.3.4/hadoop-aws-3.3.4.jar -o /opt/spark/jars/hadoop-aws-3.3.4.jar
   
   ARG spark_uid=root
   USER ${spark_uid}
   
   # install compatible numpy version
   RUN pip install numpy==1.21.6
   
   

2. Чтобы собрать образ, выполните команду:

   docker build . --tag <IMAGE-NAME>:<TAG> --platform linux/amd64
   
   

   Где:

   • <IMAGE-NAME> — имя образа.

   • <TAG> — тег образа.

3. Откройте сервис Artifact Registry.

4. Создайте репозиторий.

5. Загрузите образ.

4. Создайте задачу Managed Spark

1. Перейдите в сервис Managed Spark.

2. Откройте созданный ранее инстанс.

3. Перейдите на вкладку Задачи.

4. Нажмите Создать задачу.

5. В блоке Общие параметры введите название задачи, например spark-image-sales.

6. В блоке Образ:

   1. Выберите Пользовательский.

   2. Под полем URI образа нажмите Выбрать из реестра и выберите добавленный ранее образ.

7. В блоке Скрипт приложения:

   • В поле Тип запускаемой задачи выберите Python.

   • В поле Путь к запускаемому файлу укажите путь к файлу script-spark-image.py. В данном случае путь s3a://{bucket_name}/jobs/script-spark-image.py, где {bucket_name} — название созданного бакета Object Storage.

8. Нажмите Создать.

5. Наблюдайте за ходом выполнения задачи

Результат