Использование Apache Spark в облаке Advanced

Постановка задачи

• о фильмах

  movieId	title	genres
  1	Дьявол носит Prada (2006)	Комедия, драма
  2	Однажды в… Голливуде (2019)	Комедия, драма
  3	Шрек (2001)	Комедия, фэнтези
  4	Бойцовский клуб (1999)	Триллер, драма

• о рейтингах, которые пользователи поставили каждому фильму

  userId	movieId	rating	timestamp
  5	1	3.7	123456789
  6	2	3.6	123456789
  7	3	4.1	123456789
  8	4	4.7	123456789

Способы решения задачи

• Spark SQL

• Spark Submit

• Spark SQL без написания кода программы. Этот способ может подойти пользователям, которые не уверены в своих навыках написания кода, но знают и понимают язык SQL.

• Spark Submit с отдельно написанными короткими программами на Python. Этот способ будет ближе разработчикам.

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

1. Кластер в MapReduce Service.

2. Очередь в сервисе Data Lake Insight.

3. Бакет в объектном хранилище Object Storage Service для хранения исходных данных и сохранения результатов рассчетов.

Решение задачи в MRS с помощью Spark Submit

1. Создайте файл spark-mrs-1.py и добавьте в него следующий код:

   from pyspark.sql import SparkSession
   from pyspark.sql import functions as f
   import sys
   
   movies_path = sys.argv[1]
   ratings_path = sys.argv[2]
   save_path = sys.argv[3]
   
   spark = SparkSession.builder.appName('Movie_Rating_Above_4').getOrCreate()
   
   df_movies = spark.read.csv(movies_path, header=True, inferSchema=True)
   df_ratings = spark.read.csv(ratings_path, header=True, inferSchema=True)
   
   df_movie_rating = df_movies.join(df_ratings, "movieId", "left")
   
   df_calculated = df_movie_rating.groupBy("title").agg(f.avg("rating").alias("avg_rating"))
   
   df_calculated_above_four = df_calculated.filter("avg_rating > 4").select(["title","avg_rating"])
   
   df_calculated_above_four.write.csv(save_path, header=True)
   
   

   Этот код:

   • Принимает несколько аргументов:

     • путь к файлу с фильмами;

     • путь к файлу с рейтингами;

     • путь до папки, в которой необходимо сохранить результат.

   • Считывает данные из файлов с фильмами и рейтингами.

   • Объединяет данные в один массив по идентификатору фильма.

   • Агрегирует данные по фильмам и считает средний рейтинг (рейтинг добавляется как столбец avg_rating).

   • Фильтрует полученный массив данных.

   • Сохраняет данные в формате csv в папку.

1. Войдите в консоль управления Advanced:

   • через личный кабинет Cloud.ru;

   • как IAM-пользователь.

2. В списке сервисов выберите Object Storage Service.

3. Нажмите на созданный бакет и в левом меню выберите Objects.

4. Нажмите Upload Object.

5. Чтобы загрузить spark-mrs-1.py с кодом, в диалоговом окне нажмите Add File.

6. Нажмите Upload и дождитесь окончания загрузки файла.

1. Войдите в консоль управления Advanced:

   • через личный кабинет Cloud.ru;

   • как IAM-пользователь.

2. В списке сервисов выберите MapReduce Service.

3. В списке нажмите на название ранее созданного кластера.

4. Перейдите на вкладку Jobs и нажмите Create.

5. Задайте параметры:

   • Type — SparkSubmit.

   • Name — название задания.

   • Program Path — нажав кнопку OBS, выберите бакет, а затем загруженный ранее в него файл spark-mrs-1.py. Активируйте опцию «I confirm that the selected script is secure, and I understand the potential risks and accept the possible exceptions or impacts on the cluster» и нажмите ОК.

   • Parameters — через пробел укажите путь до файлов movies.csv, ratings.csv, а также путь до папки, куда необходимо сохранить результат (папки не должно существовать).

     Пример
     

6. Нажмите ОК.

1. Откройте бакет OBS, путь до которого был указан в процессе создания задания.

   В бакете появилась папка для сохранения результатов.

2. Нажмите на название папки.

   В ней содержится файл _SUCCESS, а также набор CSV-файлов с результатами.

3. Скачайте любой из файлов и проверьте, что результат соответствует условиям задачи.

Решение задачи в MRS с помощью Spark SQL

CREATE TEMPORARY VIEW movies (
    movieId int,
    title string,
    genres string
)
USING csv
OPTIONS (
  path 'obs://bucket-test/movies.csv',
  header true
);

CREATE TEMPORARY VIEW ratings (
    userId int,
    movieId int,
    rating float,
    timestamp string
)
USING csv
OPTIONS (
  path 'obs://bucket-test/ratings.csv',
  header true
);

INSERT OVERWRITE DIRECTORY 'obs://bucket-test/output' USING CSV
SELECT movies.title, AVG(ratings.rating) as avg_rating
FROM movies LEFT JOIN ratings ON movies.movieId = ratings.movieId
GROUP BY movies.title
HAVING avg_rating > 4

• Подключается к файлам в OBS.

• Агрегирует данные.

• Сохраняет результат в каталог на OBS.

1. Войдите в консоль управления Advanced:

   • через личный кабинет Cloud.ru;

   • как IAM-пользователь.

2. В списке сервисов выберите MapReduce Service.

3. В списке нажмите на название ранее созданного кластера.

4. Перейдите на вкладку Jobs и нажмите Create.

5. Задайте параметры:

   • Type — SparkSql.

   • Name — название задания.

   • SQL Type — Script. Введите скрипт SQL, указанный ранее.

     

6. Нажмите ОК.

1. Откройте бакет OBS, путь до которого был указан в процессе создания задания.

   В бакете появилась папка для сохранения результатов.

2. Нажмите на название папки.

   В ней содержится файл _SUCCESS, а также набор CSV-файлов с результатами.

3. Скачайте любой из файлов и проверьте, что результат соответствует условиям задачи.

Решение задачи в DLI с помощью Spark Submit

1. Войдите в консоль управления Advanced:

   • через личный кабинет Cloud.ru;

   • как IAM-пользователь.

2. В списке сервисов выберите Data Lake Insight.

3. Выберите Job Management → Spark Jobs.

4. В правом верхнему углу нажмите Create Job.

5. На вкладке Fill Form задайте параметры:

   • Queue — очередь, которая была создана в начале работы для типа задания Spark Submit.

   • Job Name — название задания.

   • Application — с помощью кнопки справа выберите созданный бакет OBS, а затем файл spark-mrs-1.py.

   • Application Parameters — на отдельных строках укажите путь до файла movies.csv, ratings.csv, а также путь до папки, куда необходимо сохранить результат (папки не должно существовать).

     Пример
     

6. В правом верхнем углу нажмите Execute.

1. Откройте бакет OBS, путь до которого был указан в процессе создания задания.

   В бакете появилась папка для сохранения результатов.

2. Нажмите на название папки.

   В ней содержится файл _SUCCESS, а также набор CSV-файлов с результатами.

3. Скачайте любой из файлов и проверьте, что результат соответствует условиям задачи.

Решение задачи в DLI с помощью Spark SQL

1. Войдите в консоль управления Advanced:

   • через личный кабинет Cloud.ru;

   • как IAM-пользователь.

2. В списке сервисов выберите Data Lake Insight.

3. В меню слева выберите SQL Editor.

4. Нажмите на вкладке Databases.

   

5. В диалоговом окне задайте параметры:

   • Database Name — название базы данных.

   • Enterprise Project — выберите необходимый проект или default.

     

6. Чтобы завершить создание базы данных, нажмите ОК.

7. Нажмите на созданную базу данных. Откроется поле Table.

   

8. Нажмите на и задайте параметры таблицы movies:

   • Name — название таблицы.

   • Data Location — OBS.

   • Data Format — CSV.

   • Bucket Location — с помощью кнопки справа выберите созданный бакет OBS, а затем файл movies.csv.

9. Опишите столбцы по примеру ниже. Добавлять столбцы можно кнопкой в правой части таблицы.

   Column Name (Название столбца)	Type (Тип данных)
   movieId	int
   title	string
   genres	string

   

10. Включите опцию Advanced settings и активируйте Table Header.

11. Нажмите ОК.

12. По аналогии создайте таблицу ratings со столбцами:

    Column Name (Название столбца)	Type (Тип данных)
    userId	int
    movieId	int
    rating	double
    timestamp	string

    

13. Нажмите ОК.

14. После того, как таблицы описаны, в правой части можно выполнить SQL запрос. Для этого вставьте запрос ниже в поле:

    INSERT OVERWRITE DIRECTORY 'obs://bucket-test/output'
    USING csv
    SELECT movies.title, AVG(ratings.rating) as avg_rating
    FROM movies LEFT JOIN ratings ON movies.movieId = ratings.movieId
    GROUP BY movies.title
    HAVING avg_rating > 4
    
    

15. В верхней части поля нажмите Execute.

(Опционально) Сборка файлов в один с помощью FunctionGraph

# -*- coding: utf-8 -*-
from com.obs.client.obs_client import ObsClient
import io
import datetime

def handler(event, context):
  date = datetime.datetime.now()
  AK = context.getAccessKey()
  SK = context.getSecretKey()
  server = 'obs.ru-moscow-1.hc.sbercloud.ru'
  bucketName = 'bucket-test'

  obsClient = ObsClient(access_key_id=AK, secret_access_key=SK, server=server)

  csv_mem = io.StringIO()

  list_objects = obsClient.listObjects(bucketName, prefix='output')
  for content in list_objects.body.contents:
    if (content.key.find('.csv') != -1):
      objectKey = str(content.key)

      resp = obsClient.getObject(bucketName, objectKey, loadStreamInMemory=True)
      if resp.body != None:
        download_object = resp.body.buffer
        csvdata = download_object.decode("utf-8")
        csv_mem.write(csvdata)

  objectKey = f"output_{date}.csv"
  obsClient.putContent(bucketName, objectKey, csv_mem.getvalue())

  return "OK"