Jalankan sesi interaktif dengan Amazon EMR Tanpa Server melalui Spark Connect

Dengan rilis Amazon EMR emr-7.13.0 dan yang lebih baru, Anda dapat terhubung ke aplikasi Amazon EMR Serverless dari PySpark klien yang dikelola sendiri seperti VS Code, PyCharm, dan notebook Jupyter menggunakan API sesi EMR Serverless dengan Apache Spark Connect. Spark Connect menggunakan arsitektur client-server yang memisahkan kode aplikasi Anda dari proses driver Spark. Anda mengembangkan dan men-debug PySpark kode di IDE lokal Anda sementara operasi Spark berjalan pada komputasi Tanpa Server EMR. Spark Connect menawarkan manfaat berikut:

Sesi Spark Connect adalah koneksi terkelola antara PySpark klien lokal Anda dan driver Spark yang berjalan di Amazon EMR Serverless. Saat Anda memulai sesi, EMR Serverless menyediakan driver Spark dan pelaksana atas nama Anda. Klien lokal Anda mengirim DataFrame dan operasi SQL ke driver, dan driver menjalankannya dari jarak jauh. Sesi berlanjut sampai Anda menghentikannya atau mencapai batas waktu idle, sehingga Anda dapat menjalankan beberapa kueri secara interaktif tanpa memulai ulang Spark. Setiap sesi memiliki URL titik akhir dan token otentikasi sendiri yang Anda gunakan untuk terhubung.

Izin yang diperlukan

Selain izin yang diperlukan untuk mengakses Amazon EMR Serverless, tambahkan juga izin berikut ke peran IAM Anda untuk mengakses titik akhir Spark Connect dan mengelola sesi Spark Connect:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "EMRServerlessApplicationLevelAccess",
            "Effect": "Allow",
            "Action": [
                "emr-serverless:StartSession",
                "emr-serverless:ListSessions"
            ],
            "Resource": [
                "arn:aws:emr-serverless:us-east-1:111122223333:/applications/application-id"
            ]
        },
        {
            "Sid": "EMRServerlessSessionLevelAccess",
            "Effect": "Allow",
            "Action": [
                "emr-serverless:GetSession",
                "emr-serverless:GetSessionEndpoint",
                "emr-serverless:TerminateSession",
                "emr-serverless:GetResourceDashboard"
            ],
            "Resource": [
                "arn:aws:emr-serverless:us-east-1:111122223333:/applications/application-id/sessions/*"
            ]
        },
        {
            "Sid": "EMRServerlessRuntimeRoleAccess",
            "Effect": "Allow",
            "Action": [
                "iam:PassRole"
            ],
            "Resource": [
                "arn:aws:iam::111122223333:role/EMRServerlessExecutionRole"
            ],
            "Condition": {
                "StringLike": {
                    "iam:PassedToService": "emr-serverless.amazonaws.com"
                }
            }
        }
    ]
}

Bekerja dengan sesi interaktif

Untuk membuat Connect-enabled aplikasi Spark dan menghubungkannya, ikuti langkah-langkah ini.

Berikut ini adalah contoh skrip Python untuk memulai sesi dan mengirim permintaan langsung ke titik akhir sesi:

import boto3
import time
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

client = boto3.client('emr-serverless', region_name='REGION')

APPLICATION_ID = 'APPLICATION_ID'
EXECUTION_ROLE = 'arn:aws:iam::account-id:role/EMRServerlessExecutionRole'

# Start the session
response = client.start_session(
    applicationId=APPLICATION_ID,
    executionRoleArn=EXECUTION_ROLE
)
session_id = response['sessionId']
print(f"Session {session_id} starting...")

# Wait for the session to be ready
while True:
    response = client.get_session(
        applicationId=APPLICATION_ID,
        sessionId=session_id
    )
    state = response['session']['state']
    print(f"Session state: {state}")
    if state in ('STARTED', 'IDLE'):
        break
    if state in ('FAILED', 'TERMINATED'):
        raise Exception(f"Session failed: {response['session'].get('stateDetails', 'Unknown error')}")
    time.sleep(5)

# Retrieve the Spark Connect endpoint and authentication token
response = client.get_session_endpoint(
    applicationId=APPLICATION_ID,
    sessionId=session_id
)

# Construct the authenticated remote URL
auth_token = response['authToken']
endpoint_url = response['endpoint']
connect_url = endpoint_url.replace("https://", "sc://", 1) + ":443/;use_ssl=true;"
connect_url += f"x-aws-proxy-auth={auth_token}"

# Start the Spark session
spark = SparkSession.builder.remote(connect_url).getOrCreate()
print(f"Connected. Spark version: {spark.version}")

# Run SQL
spark.sql("SELECT 1+1 AS result").show()

# Run DataFrame operations
df = spark.range(100).withColumn("squared", col("id") * col("id"))
df.show(10)
print(f"Count: {df.count()}")

# Stop the Spark session (disconnects the client only)
spark.stop()

# Terminate the EMR Serverless session to stop billing.
# spark.stop() only closes the local client connection. The remote session
# continues running and incurring charges until you explicitly terminate it
# or it reaches the idle timeout.
client.terminate_session(
    applicationId=APPLICATION_ID,
    sessionId=session_id
)
print(f"Session {session_id} terminated.")

Untuk mengakses UI Spark langsung atau Server Sejarah Spark untuk sesi, gunakan GetResourceDashboard API.

response = client.get_resource_dashboard(
    applicationId=APPLICATION_ID,
    resourceId=session_id,
    resourceType='SESSION'
)
response['url']

Saat sesi aktif, URL membuka UI Apache Spark langsung untuk pemantauan kueri, tahapan, dan pelaksana secara real-time. Setelah sesi berakhir, Server Sejarah Spark tetap tersedia untuk analisis pasca-sesi melalui konsol Amazon EMR Serverless.

Pertimbangan dan batasan

Pertimbangkan hal berikut saat menjalankan beban kerja interaktif melalui Spark Connect.