Esegui sessioni interattive con Amazon EMR Serverless tramite Spark Connect

Con la versione di Amazon EMR emr-7.13.0 e successive, puoi connetterti a un'applicazione Amazon EMR Serverless da PySpark client autogestiti come VS Code e notebook Jupyter utilizzando la sessione EMR Serverless con Apache Spark Connect. PyCharm APIs Spark Connect utilizza un'architettura client-server che disaccoppia il codice dell'applicazione dal processo del driver Spark. Sviluppi ed esegui il debug PySpark del codice nel tuo IDE locale mentre le operazioni Spark vengono eseguite sull'elaborazione EMR Serverless. Spark Connect offre i seguenti vantaggi:

Una sessione Spark Connect è una connessione gestita tra il PySpark client locale e un driver Spark in esecuzione su Amazon EMR Serverless. Quando avvii una sessione, EMR Serverless effettua il provisioning di un driver Spark e di esecutori per tuo conto. Il client locale invia DataFrame le operazioni SQL al driver, che le esegue in remoto. La sessione persiste finché non viene terminata o non raggiunge il timeout di inattività, quindi puoi eseguire più query in modo interattivo senza riavviare Spark. Ogni sessione ha il proprio URL dell'endpoint e il token di autenticazione che usi per connetterti.

Autorizzazioni richieste

Oltre alle autorizzazioni necessarie per accedere ad Amazon EMR Serverless, aggiungi anche le seguenti autorizzazioni al tuo ruolo IAM per accedere a un endpoint Spark Connect e gestire le sessioni Spark Connect:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "EMRServerlessApplicationLevelAccess",
            "Effect": "Allow",
            "Action": [
                "emr-serverless:StartSession",
                "emr-serverless:ListSessions"
            ],
            "Resource": [
                "arn:aws:emr-serverless:region:account-id:/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:region:account-id:/applications/application-id/sessions/*"
            ]
        },
        {
            "Sid": "EMRServerlessRuntimeRoleAccess",
            "Effect": "Allow",
            "Action": [
                "iam:PassRole"
            ],
            "Resource": [
                "arn:aws:iam::account-id:role/EMRServerlessExecutionRole"
            ],
            "Condition": {
                "StringLike": {
                    "iam:PassedToService": "emr-serverless.amazonaws.com"
                }
            }
        }
    ]
}

Lavorare con sessioni interattive

Per creare un'applicazione compatibile con Spark Connect e connetterti ad essa, segui questi passaggi.

Di seguito è riportato un esempio di script Python per avviare una sessione e inviare richieste direttamente all'endpoint della sessione:

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

Per accedere all'interfaccia utente live di Spark o allo Spark History Server per una sessione, usa l'API. GetResourceDashboard

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

Mentre una sessione è attiva, l'URL apre l'interfaccia utente live di Apache Spark per il monitoraggio in tempo reale di query, fasi ed esecutori. Al termine di una sessione, Spark History Server rimane disponibile per l'analisi post-sessione tramite la console Amazon EMR Serverless.

Considerazioni e limitazioni

Considera quanto segue quando esegui carichi di lavoro interattivi tramite Spark Connect.