Execute sessões interativas com o Amazon EMR Serverless por meio do Spark Connect

Com a versão do Amazon EMR emr-7.13.0 e posterior, você pode se conectar a um aplicativo Amazon EMR Serverless a partir de PySpark clientes autogerenciados, como o VS Code e os notebooks Jupyter PyCharm, usando as APIs de sessão sem servidor do EMR com o Apache Spark Connect. O Spark Connect usa uma arquitetura cliente-servidor que separa o código do aplicativo do processo do driver do Spark. Você desenvolve e depura o PySpark código em seu IDE local enquanto as operações do Spark são executadas na computação EMR Serverless. O Spark Connect oferece os seguintes benefícios:

Uma sessão do Spark Connect é uma conexão gerenciada entre seu PySpark cliente local e um driver Spark em execução no Amazon EMR Serverless. Quando você inicia uma sessão, o EMR Serverless provisiona um driver e executores do Spark em seu nome. Seu cliente local envia DataFrame operações SQL para o driver, e o driver as executa remotamente. A sessão persiste até que você a encerre ou atinja o tempo limite de inatividade, para que você possa executar várias consultas interativamente sem reiniciar o Spark. Cada sessão tem seu próprio URL de endpoint e token de autenticação que você usa para se conectar.

Permissões obrigatórias

Além das permissões necessárias para acessar o Amazon EMR Serverless, adicione também as seguintes permissões à sua função do IAM para acessar um endpoint do Spark Connect e gerenciar sessões do 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"
                }
            }
        }
    ]
}

Trabalhando com sessões interativas

Para criar um Connect-enabled aplicativo Spark e conectar-se a ele, siga estas etapas.

Veja a seguir um exemplo de script Python para iniciar uma sessão e enviar solicitações diretamente para o endpoint da sessão:

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 acessar a interface de usuário do Spark ao vivo ou o Spark History Server para uma sessão, use a GetResourceDashboard API.

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

Enquanto uma sessão está ativa, a URL abre a interface de usuário ativa do Apache Spark para monitoramento em tempo real de consultas, estágios e executores. Após o término de uma sessão, o Spark History Server permanece disponível para análise pós-sessão por meio do console Amazon EMR Serverless.

Considerações e limitações

Considere o seguinte ao executar cargas de trabalho interativas por meio do Spark Connect.