E-MapReduce:Use third-party Python libraries in a PySpark job

Go to the Runtime Environment Management page.

1. Log on to the E-MapReduce console.

2. In the navigation pane on the left, choose EMR Serverless > Spark.

3. On the Spark page, click the name of the target workspace.

4. On the EMR Serverless Spark page, click Environment in the navigation pane on the left.

Click Create Environment.

On the Create Environment page, click Add Library.

For more information about parameters, see Manage runtime environments.

In the New Library dialog box, use the PyPI source type, configure the PyPI Package parameter, and then click OK.

In the PyPI Package field, enter the name and version of the library. If you do not specify a version, the latest version is installed by default.

In this example, the libraries faker and geopy are added.

Click Create.

After the runtime environment is created, the system starts to initialize the environment.

Click pyspark_third_party_libs_demo.py to download the required resource file.

This example demonstrates how to use PySpark and third-party libraries to generate analog data and perform geographic analysis. The faker library is used to generate analog data that contains user information and random geographic locations. The geopy library is used to calculate the geographic distance between each user's location and the Eiffel Tower. Finally, users within a 10-kilometer range are filtered out.

You can also create the pyspark_third_party_libs_demo.py sample script with the following content:

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()

# Use the third-party faker library to generate analog data
fake = Faker()
landmark = (48.8584, 2.2945)  # Eiffel Tower coordinates

# Create a function to generate analog data
def generate_fake_data(num_records):
    data = []
    for _ in range(num_records):
        # Generate random coordinates near Paris
        lat = 48.85 + random.uniform(-0.2, 0.2)
        lon = 2.30 + random.uniform(-0.2, 0.2)
        data.append((
            fake.uuid4(),        # User ID
            fake.name(),         # Name
            lat,                 # Latitude
            lon                  # Longitude
        ))
    return data

# Generate 100 analog records
fake_data = generate_fake_data(100)

# Create a Spark DataFrame
columns = ["user_id", "name", "latitude", "longitude"]
df = spark.createDataFrame(fake_data, schema=columns)

# Print the generated sample data
print("Generated sample data:")
df.show(5)

# Use the third-party geopy library to calculate distance
def calculate_distance(lat, lon, landmark=landmark):
    """Calculate the geographic distance between two points (kilometers)"""
    user_location = (lat, lon)
    return geodesic(user_location, landmark).kilometers

# Register a UDF (User Defined Function)
distance_udf = udf(calculate_distance, FloatType())

# Add a distance column
df_with_distance = df.withColumn(
    "distance_km", 
    distance_udf("latitude", "longitude")
)

# Find users within a 10-kilometer range
nearby_users = df_with_distance.filter("distance_km <= 10")

# Print the results
print(f"\nFound {nearby_users.count()} users within a 10-kilometer range:")
nearby_users.select("name", "latitude", "longitude", "distance_km").show(10)

Upload pyspark_third_party_libs_demo.py to OSS. For more information, see Simple upload.

In the navigation pane on the left of the EMR Serverless Spark page, click Development.

On the Development tab, click the icon.

In the New dialog box, enter a name, select Application(Batch) > PySpark from the Type drop-down list, and then click OK.

In the upper-right corner, select a queue.

On the configuration tab of the job, configure the following parameters and click Run. You do not need to configure other parameters.

After the job is run, click Logs in the Actions column of the job in the Execution Records section.

On the Log Exploration tab, you can view the related logs.

For example, on the Stdout tab of the Driver Log section, you can view the following information:

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

Create an Elastic Compute Service (ECS) instance that uses the Alibaba Cloud Linux 3 OS, is connected to the Internet, and has an x86 architecture. For more information, see Create an instance on the Custom Launch tab.

Run the following commands to install 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

Build a Conda environment that uses Python 3.8 and NumPy.

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

Click kmeans.py and kmeans_data.txt to download the required resource files.

You can also create the kmeans.py sample script and the kmeans_data.txt data file with the following content:

"""
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

Upload the pyspark_conda_env.tar.gz, kmeans.py, and kmeans_data.txt files to OSS. For more information, see Simple upload.

In the navigation pane on the left of the EMR Serverless Spark page, click Development.

On the Development tab, click the icon.

In the New dialog box, enter a name, select Application(Batch) > PySpark from the Type drop-down list, and then click OK.

In the upper-right corner, select a queue.

On the configuration tab of the job, configure the following parameters and click Run. You do not need to configure other parameters.

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

After the job is run, click Logs in the Actions column of the job in the Execution Records section.

On the Log Exploration tab, you can view the related logs.

For example, on the Stdout tab of the Driver Log section, you can view the following information:

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