Realice sesiones interactivas con Amazon EMR Serverless a través de Spark Connect

Con la versión Amazon EMR emr-7.13.0 y versiones posteriores, puede conectarse a una aplicación Amazon EMR Serverless desde PySpark clientes autogestionados como VS Code y Jupyter notebooks mediante la sesión PyCharm EMR Serverless con Apache Spark Connect. APIs Spark Connect utiliza una arquitectura cliente-servidor que desacopla el código de la aplicación del proceso del controlador de Spark. Desarrollas y depura el PySpark código en tu IDE local mientras las operaciones de Spark se ejecutan en el procesamiento EMR Serverless. Spark Connect ofrece las siguientes ventajas:

Una sesión de Spark Connect es una conexión gestionada entre tu PySpark cliente local y un controlador de Spark que se ejecuta en Amazon EMR Serverless. Cuando inicias una sesión, EMR Serverless aprovisiona un controlador Spark y ejecutores en tu nombre. Su cliente local envía DataFrame las operaciones de SQL al controlador, que las ejecuta de forma remota. La sesión persiste hasta que la termines o se agote el tiempo de espera de inactividad, por lo que puedes ejecutar varias consultas de forma interactiva sin necesidad de reiniciar Spark. Cada sesión tiene su propia URL de punto final y token de autenticación que utilizas para conectarte.

Permisos necesarios

Además de los permisos necesarios para acceder a Amazon EMR Serverless, añada también los siguientes permisos a su función de IAM para acceder a un punto final de Spark Connect y gestionar las sesiones de 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"
                }
            }
        }
    ]
}

Trabajando con sesiones interactivas

Para crear una aplicación compatible con Spark Connect y conectarte a ella, sigue estos pasos.

El siguiente es un ejemplo de script de Python para iniciar una sesión y enviar solicitudes directamente al punto final de la sesión:

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

Para acceder a la interfaz de usuario activa de Spark o al servidor de historial de Spark durante una sesión, usa la GetResourceDashboard API.

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

Mientras una sesión está activa, la URL abre la interfaz de usuario activa de Apache Spark para monitorizar en tiempo real las consultas, las etapas y los ejecutores. Una vez finalizada la sesión, el servidor de historial de Spark permanece disponible para su análisis posterior a la sesión a través de la consola Amazon EMR Serverless.

Consideraciones y limitaciones

Ten en cuenta lo siguiente cuando ejecutes cargas de trabajo interactivas a través de Spark Connect.