Esegui sessioni interattive con Amazon EMR Serverless tramite Spark Connect

Con la versione 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 le API di PyCharm sessione EMR Serverless con Apache Spark Connect. Spark Connect utilizza un'architettura client-server che separa il codice dell'applicazione dal processo del driver Spark. Sviluppi ed esegui il debug del PySpark codice nel tuo IDE locale mentre le operazioni Spark vengono eseguite su EMR Serverless Computing. 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 esegue il provisioning di un driver Spark e degli esecutori per tuo conto. Il client locale invia DataFrame le operazioni SQL al driver, che le esegue in remoto. La sessione persiste finché non la interrompi 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 richieste 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: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"
                }
            }
        }
    ]
}

Lavorare con sessioni interattive

Per creare un' Connect-enabled applicazione Spark 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 attiva 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 attiva di Apache Spark per il monitoraggio in tempo reale di query, fasi ed esecutori. Al termine di una sessione, lo Spark History Server rimane disponibile per l'analisi post-sessione tramite la console Amazon EMR Serverless.

Considerazioni e limitazioni

Quando esegui carichi di lavoro interattivi tramite Spark Connect, considera quanto segue.