E-MapReduce:PySpark ジョブでサードパーティの Python ライブラリを使用する

ランタイム環境管理ページに移動します。

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

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

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

4. [EMR Serverless Spark] ページで、左側のナビゲーションウィンドウにある [ランタイム環境管理] をクリックします。

[ランタイム環境の作成] をクリックします。

[ランタイム環境の作成] ページで、[ライブラリの追加] をクリックします。

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

[新しいライブラリ] ダイアログボックスで、[PyPI] ソースタイプを使用し、[PyPI パッケージ] パラメーターを設定してから、[OK] をクリックします。

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

この例では、ライブラリ faker と geopy が追加されます。

[作成] をクリックします。

ランタイム環境が作成されると、システムは環境の初期化を開始します。

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 キロメートル以内のユーザーを検索する
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 ページの左側のナビゲーションウィンドウで、[データ開発] をクリックします。

[開発] タブで、 アイコンをクリックします。

[作成] ダイアログボックスで、名前を入力し、[タイプ] ドロップダウンリストから [バッチジョブ] > [PySpark] を選択して、[OK] をクリックします。

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

ジョブの設定タブで、次のパラメーターを設定し、[実行] をクリックします。他のパラメーターを設定する必要はありません。

ジョブの実行後、[実行記録] セクションのジョブの [アクション] 列で [ログエクスプローラー] をクリックします。

[ログエクスプローラー] タブで、関連するログを表示できます。

たとえば、[ドライバーログ] セクションの [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 OS を使用し、インターネットに接続され、x86 アーキテクチャを持つ Elastic Compute Service (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 データファイルを作成することもできます。

"""
MLlib を使用した K-means クラスタリングプログラム。

この例には 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 ページの左側のナビゲーションウィンドウで、[データ開発] をクリックします。

[開発] タブで、 アイコンをクリックします。

[作成] ダイアログボックスで、名前を入力し、[タイプ] ドロップダウンリストから [バッチジョブ] > [PySpark] を選択して、[OK] をクリックします。

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

ジョブの設定タブで、次のパラメーターを設定し、[実行] をクリックします。他のパラメーターを設定する必要はありません。

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

ジョブの実行後、[実行記録] セクションのジョブの [アクション] 列で [ログエクスプローラー] をクリックします。

[ログエクスプローラー] タブで、関連するログを表示できます。

たとえば、[ドライバーログ] セクションの [Stdout] タブで、次の情報を表示できます。

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