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

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

Una sesión de Spark Connect es una conexión gestionada entre su PySpark cliente local y un controlador de Spark que se ejecuta en Amazon EMR Serverless. Cuando inicia una sesión, EMR Serverless aprovisiona un controlador de Spark y ejecutores en su nombre. Tu cliente local envía DataFrame las operaciones de SQL al controlador, y el controlador las ejecuta de forma remota. La sesión continúa hasta que la finalices o hasta que se agote el tiempo de inactividad, por lo que puedes ejecutar varias consultas de forma interactiva sin tener que reiniciar Spark. Cada sesión tiene su propia URL de punto final y un token de autenticación que puedes usar 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: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"
                }
            }
        }
    ]
}

Trabajando con sesiones interactivas

Para crear una Connect-enabled aplicación de Spark y conectarte a ella, sigue estos pasos.

El siguiente es un ejemplo de un 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 de Spark en vivo 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 de Apache Spark en vivo para monitorear en tiempo real las consultas, las etapas y los ejecutores. Una vez finalizada una 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 sin servidor de Amazon EMR.

Consideraciones y limitaciones

Tenga en cuenta lo siguiente al ejecutar cargas de trabajo interactivas a través de Spark Connect.