E-MapReduce:PySpark プログラムでサードパーティの Python ライブラリを使用する

環境ページに移動します。

1. E-MapReduce コンソールにログインします。

2. 左側のナビゲーションウィンドウで、EMR Serverless > Spark を選択します。

3. Spark ページで、対象のワークスペース名をクリックします。

4. EMR Serverless Spark ページの左ナビゲーションバーで、Environment を選択します。

Create Environment をクリックします。

Create Environment ページで、Add Library をクリックします。

パラメーターの詳細については、「実行環境の管理」をご参照ください。

New Library ダイアログボックスで、ソースタイプを PyPI に設定し、PyPI Package パラメーターを構成してから、OK をクリックします。

PyPI Package フィールドには、ライブラリ名とバージョンを入力します。バージョンを指定しない場合、デフォルトで最新バージョンがインストールされます。

この例では、faker および geopy ライブラリを追加します。

create をクリックします。

[作成] をクリックすると、環境の初期化が開始されます。

pyspark_third_party_libs_demo.py をクリックして、必要なリソースファイルをダウンロードします。

この例では、サードパーティのライブラリを使用した PySpark により合成データを生成し、地理空間分析を実行します。faker ライブラリはランダムな位置情報を含む合成ユーザーデータを生成し、geopy ライブラリは各位置とエッフェル塔との距離を計算して、10 キロメートル圏内にいるユーザーを特定します。

以下の内容で pyspark_third_party_libs_demo.py というサンプルスクリプトを作成することもできます。

from pyspark.sql import SparkSession   
from pyspark.sql.functions import udf
from pyspark.sql.types import FloatType
from faker import Faker
import random
from geopy.distance import geodesic

spark = SparkSession.builder \
        .appName("PySparkThirdPartyLibsDemo") \
        .getOrCreate()

# サードパーティの Faker ライブラリを使用して合成データを生成します。
fake = Faker()
landmark = (48.8584, 2.2945)  # エッフェル塔の座標

# 合成データを生成する関数を作成します。
def generate_fake_data(num_records):
    data = []
    for _ in range(num_records):
        # パリ近郊のランダムな座標を生成します。
        lat = 48.85 + random.uniform(-0.2, 0.2)
        lon = 2.30 + random.uniform(-0.2, 0.2)
        data.append((
            fake.uuid4(),        # ユーザー ID
            fake.name(),         # 氏名
            lat,                 # 緯度
            lon                  # 経度
        ))
    return data

# 100 件の合成レコードを生成します。
fake_data = generate_fake_data(100)

# Spark DataFrame を作成します。
columns = ["user_id", "name", "latitude", "longitude"]
df = spark.createDataFrame(fake_data, schema=columns)

# 生成されたサンプルデータを出力します。
print("Generated sample data:")
df.show(5)

# サードパーティの geopy ライブラリを使用して距離を計算します。
def calculate_distance(lat, lon, landmark=landmark):
    """2 点間の地理的距離をキロメートル単位で計算します。"""
    user_location = (lat, lon)
    return geodesic(user_location, landmark).kilometers

# ユーザー定義関数 (UDF) を登録します。
distance_udf = udf(calculate_distance, FloatType())

# 距離カラムを追加します。
df_with_distance = df.withColumn(
    "distance_km", 
    distance_udf("latitude", "longitude")
)

# 10 km 圏内のユーザーを抽出します。
nearby_users = df_with_distance.filter("distance_km <= 10")

# 結果を出力します。
print(f"\nFound {nearby_users.count()} users within a 10-kilometer range:")
nearby_users.select("name", "latitude", "longitude", "distance_km").show(10)

pyspark_third_party_libs_demo.py を OSS にアップロードします。手順については、「シンプルアップロード」をご参照ください。

EMR Serverless Spark ページの左側ナビゲーションウィンドウで、Development をクリックします。

Development タブで、 アイコンをクリックします。

Create ダイアログボックスで、名前を入力し、タイプとして Application > PySpark を選択してから、OK をクリックします。

右上隅でキューを選択します。

新しい開発タブで、以下のパラメーターを構成し、他のパラメーターはデフォルト設定のままにしてから、実行 をクリックします。

タスク実行後、Execution Records セクションに移動し、タスクの [操作] 列で Log Exploration をクリックします。

Log Exploration タブで、ログ情報を確認します。

たとえば、Driver Log の Stdout タブでは、次のような情報が表示されます。

Generated sample data:
+--------------------+-------------------+------------------+------------------+
|             user_id|               name|          latitude|         longitude|
+--------------------+-------------------+------------------+------------------+
|73d4565c-8cdf-4bc...|  Garrett Robertson| 48.81845614776422|2.4087517234236064|
|0fc364b1-6759-416...|      Dawn Gonzalez| 48.68654896170054|2.4708555780468013|
|2ab1f0aa-5552-4e1...|Alexander Gallagher| 48.87603770688707|2.1209399987431246|
|1cabbdde-e703-4a8...|       David Morris|48.656356532418116|2.2503952330408175|
|8b7938a0-b283-401...|    Shannon Perkins| 48.82915001905855| 2.410743969589327|
+--------------------+-------------------+------------------+------------------+
only showing top 5 rows


Found 24 users within a 10-kilometer range:
+-----------------+------------------+------------------+-----------+
|             name|          latitude|         longitude|distance_km|
+-----------------+------------------+------------------+-----------+
|Garrett Robertson| 48.81845614776422|2.4087517234236064|   9.490705|
|  Shannon Perkins| 48.82915001905855| 2.410743969589327|   9.131355|
|      Alex Harris| 48.82547383207313|2.3579336032430027|   5.923493|
|      Tammy Ramos| 48.84668267431606|2.3606455536493574|   5.026109|
|   Ivan Christian| 48.89224239228342|2.2811025348668195|  3.8897192|
|  Vernon Humphrey| 48.93142188723839| 2.306957802222233|   8.171813|
|  Shawn Rodriguez|48.919907710882654|2.2270993307836044|   8.439087|
|    Robert Fisher|48.794216103154646|2.3699024070507906|   9.033209|
|  Heather Collier|48.822957591865205|2.2993033803043454|   3.957171|
|       Dawn White|48.877816307255586|2.3743880390928878|   6.246059|
+-----------------+------------------+------------------+-----------+
only showing top 10 rows

Alibaba Cloud Linux 3 を実行し、パブリック IP アドレスを持ち、x86 アーキテクチャを使用する ECS インスタンスを作成します。詳細については、「ウィザードを使用したインスタンスの作成」をご参照ください。

以下のコマンドを実行して Miniconda をインストールします。

wget https://repo.continuum.io/miniconda/Miniconda3-latest-Linux-x86_64.sh
chmod +x Miniconda3-latest-Linux-x86_64.sh

./Miniconda3-latest-Linux-x86_64.sh -b
source miniconda3/bin/activate

Python 3.8 と NumPy を使用する Conda 環境を構築します。

conda create -y -n pyspark_conda_env -c conda-forge conda-pack numpy python=3.8
conda activate pyspark_conda_env
conda pack -f -o pyspark_conda_env.tar.gz

kmeans.py および kmeans_data.txt をクリックして、必要なリソースファイルをダウンロードします。

以下の内容でサンプルスクリプト kmeans.py およびデータファイル kmeans_data.txt を作成することもできます。

"""
A K-means clustering program using MLlib.

This example requires NumPy (http://www.numpy.org/).
"""
import sys

import numpy as np
from pyspark import SparkContext
from pyspark.mllib.clustering import KMeans


def parseVector(line):
    return np.array([float(x) for x in line.split(' ')])


if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: kmeans <file> <k>", file=sys.stderr)
        sys.exit(-1)
    sc = SparkContext(appName="KMeans")
    lines = sc.textFile(sys.argv[1])
    data = lines.map(parseVector)
    k = int(sys.argv[2])
    model = KMeans.train(data, k)
    print("Final centers: " + str(model.clusterCenters))
    print("Total Cost: " + str(model.computeCost(data)))
    sc.stop()

0.0 0.0 0.0
0.1 0.1 0.1
0.2 0.2 0.2
9.0 9.0 9.0
9.1 9.1 9.1
9.2 9.2 9.2

pyspark_conda_env.tar.gz、kmeans.py、および kmeans_data.txt を OSS にアップロードします。手順については、「シンプルアップロード」をご参照ください。

EMR Serverless Spark ページの左側ナビゲーションウィンドウで、Development をクリックします。

Development タブで、 アイコンをクリックします。

Create ダイアログボックスで、名前を入力し、タイプとして Application > PySpark を選択してから、OK をクリックします。

右上隅でキューを選択します。

新しい開発タブで、以下のパラメーターを構成し、他のパラメーターはデフォルト設定のままにしてから、実行 をクリックします。

spark.pyspark.driver.python  ./condaenv/bin/python
spark.pyspark.python         ./condaenv/bin/python

タスク実行後、下部の Execution Records エリアで、タスクの [操作] 列にある Log Exploration をクリックします。

Log Exploration タブで、ログ情報を確認します。

たとえば、Driver Log の Stdout タブでは、次のような出力が表示されます。

Final centers: [array([0.1, 0.1, 0.1]), array([9.1, 9.1, 9.1])]
Total Cost: 0.11999999999999958