トップ > re:Invent 2022 で発表された AWS Glue for Rayを使ってみた！

re:Invent 2022 で発表された AWS Glue for Rayを使ってみた！

記事タイトルとURLをコピーする

本記事は2022/12/14時点の情報です。

サーバーワークス　DS1課の吉岡です。
re:Invent 2022においてAWS Glue for Rayが発表されました。
この記事ではAWS Glue for Rayの使い方についてまとめたいと思います。

aws.amazon.com

はじめに

AWS Glue Studioで選択できるジョブとしてRayが選択できるようになりました。
このRayの使い方を説明します。

aws.amazon.com

プレビュー版環境

このアップデートはプレビュー段階のため、一部のリージョンでのみ使用できます。

  • バージニア北部リージョン(us-east-1)
  • オハイオリージョン(us-east-2)
  • オレゴンリージョン(us-west-2)
  • 東京リージョン(ap-northease-1)
  • アイルランドリージョン(eu-west-1)

Rayとは

RayはPythonのライブラリの一つで、シンプルに並列分散処理を記述できます。 docs.ray.io

並列分散処理とは

データに対する処理をより高速にするために並列処理を行うことです。
複数台のコンピューターを同時に使って、一つの計算処理を行うようなイメージです。 コンピューターが1台の時と比較して高速に処理できます。

今回のアップデートでAmazon GlueのETL処理をRayを用いて並列分散処理できます。

SparkとRay

これまでは分散並列処理を行うジョブとしてSparkを選択することができました。　

どちらも分散並列処理を行うことができますが以下のような違いがあります。

Sparkの特徴

  • 機械学習，ストリーミング，複雑なクエリなど幅広い領域を簡単に表現できる
  • Hadoopなどの他のビックデータのツールとの組み合わせが可能である

Rayの特徴　

  • Pythonicなコードを実装できる（Pythonのライブラリと組み合わせられる）

RayはPythonとの相性がよく、シンプルなコードを書くことができます。
違いを説明しましたが、分散処理のフレームワークはユースケースにより様々な選択があります。

実際にやってみた

実際に使用した流れをコンソール画面と共に説明します。

コード内容

今回の実装内容で使うRayライブラリを用いたコードは以下になります。
公式サイトのサンプルコードを用いています。

サンプルコード概要

このサンプルコードはモンテカルロ法を用いて円周率の近似値を求めています。

  • 正方形の中でランダムに座標(x,y)を取ります。
  • 座標が円の中にあるかどうか判定します。(x2+y2<1であれば座標は円の中に存在する)
  • この判定を何回も繰り返し、
    π={（座標が円の中に存在した回数）×4} / { 繰り返した総数 }
    によって、円周率の近似値が求められます。

この繰り返しの総数が多ければ多いほど、より正確な円周率を求めることができます。
この繰り返し処理にRayを用いています。

コードの中身

サンプルコードにコードの説明を付け足しています。

# 座標(x,y)にランダムの数字を代入する
# その座標が円の中に存在するかどうかを判定する
import ray
from random import random

# rayを使いますよの宣言のようなものです
ray.init()

# SAMPLESが関数の中で繰り返し行う回数
SAMPLES = 1000000; 

# @ray.remoteを付け加えることでRay remote関数が使用できる
# この関数の中の処理を並列処理することができます
@ray.remote
def pi4_sample():
    in_count = 0
    for _ in range(SAMPLES):
        x, y = random(), random()
        if x*x + y*y <= 1:
            in_count += 1
    return in_count

future = pi4_sample.remote()
pi = ray.get(future) * 4.0 / SAMPLES
print(f'{pi} is an approximation of pi') 

# 上記の関数pi4_sampleを100000回行いたいです。
# 直列処理だと処理に11時間はかかりますが、Rayを扱うことで2時間になります。
# 10ノードのRayクラスタなら約10分 に短縮できます

BATCHES = 100000
#resulsには関数実行毎の円内に存在した回数が追加されます
results = [] 
for _ in range(BATCHES):
    results.append(pi4_sample.remote())
output = ray.get(results)
pi = sum(output) * 4.0 / BATCHES / SAMPLES
print(f'{pi} is a way better approximation of pi')

www.ray.io

Rayを用いてのジョブ作成

ジョブ作成時には、実行するファイルを新規作成 or ローカルからアップロード を選択できます。
今回はファイルを新規作成を選択後、ジョブの作成に進みます。

スクリプトの書き込み

作成ボタンをクリックすると、スクリプトのページに遷移します。
このページでRayライブラリを用いての実装内容を記述します。
今回は先ほど説明したサンプルコードを記述しています。

ジョブの詳細設定

実装ができると、ジョブの詳細を設定します。
この時にジョブに付与するIAMロールが必要になります。
Amazon Glueでは先ほど実装したファイルをS3バケットに保存する仕様になっています。 ジョブの実行時にS3のファイルを取りに行くためのロールが必要です。
IAMロールに関してAWS公式の参考ページがありました。

docs.aws.amazon.com

保存と実行

最後にジョブの保存と実行のボタンを押して、ジョブが実行します。

実行内容の確認

今回はジョブが終了するまで約20分ほどかかりました。
実行結果はCloud Watch のロググループのページで確認できました。
円周率を求めることができています。

Spark vs Ray

ジョブのワーカータイプを同じ条件で、
同じ分散並列処理をSparkとRayで行い、違いがあるのか確認しました。
先ほどの円周率を求める分散処理をPySparkライブラリを用いて実装しました。

コードの中身

10億回の処理を分散並列処理しています。

  • Ray を用いたコード
import ray
from random import random

ray.init()

SAMPLES = 1000000000; 
@ray.remote
def pi4_sample():
    in_count = 0
    for _ in range(SAMPLES):
        x, y = random(), random()
        if x*x + y*y <= 1:
            in_count += 1
    return in_count

future = pi4_sample.remote()
pi = ray.get(future) * 4.0 / SAMPLES
print(f'{pi} is an approximation of pi') 

BATCHES = 1
results = [] 
for _ in range(BATCHES):
    results.append(pi4_sample.remote())
output = ray.get(results)
pi = sum(output) * 4.0 / BATCHES / SAMPLES
print(f'{pi} is a way better approximation of pi')
  • Sparkを用いたコード
from time import time
import numpy as np
from random import random
from operator import add

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('CalculatePi').getOrCreate()
sc = spark.sparkContext

n = 1000000000

def is_point_inside_unit_circle(p):
    x, y = random(), random()
    return 1 if x*x + y*y < 1 else 0

t_0 = time()

count = sc.parallelize(range(0, n)) \
             .map(is_point_inside_unit_circle).reduce(add)
print(np.round(time()-t_0, 3), "seconds elapsed for spark approach and n=", n)
print("Pi is roughly %f" % (4.0 * count / n))

コードを書いてみて

Rayだと@ray.remoteの関数が分散並列処理される仕様になっています。
Sparkではメソッドで分散並列処理を記述します。
@ray.remoteがあることで、分散並列処理が行われている箇所を一目で判断できます。
見やすさという点ではRayのほうが見やすいコードを記述できると感じました。

実行結果

結果として処理時間がRayでは11分、Sparkでは3分になりました。
この比較ではSparkのほうが処理を速く行うことができるという結果になりました。
(実行時間に差があり個人的には驚きです。。。Sparkも凄いです)

使ってみて

今回はAWS Glue for Rayを使ってみました。
コンソール画面で簡単にジョブを作成できたので驚きました。（嬉しい驚きです）
普段 Rayを用いて分散並列処理を行っている方にとって嬉しいアップデートだったのではないでしょうか。
良ければこの記事を参考にして、活用していただけたらと思います。

▼AWS re:Invent 2022▼
米・ラスベガスで開催されるAWS最大のカンファレンスイベント

最後まで読んでいただき、ありがとうございました。