通过 Spark Connect 与亚马逊 EMR Serverless 进行交互式会话

在 Amazon EMR 版本emr-7.13.0及更高版本中，您可以使用与 Apache Spark Connect 的 EMR 无服务器会话从 VS Code 等自我管理的 PySpark 客户端（例如 VS Code）和 Jupyter 笔记本连接到 Amazon EMR 无服务器应用程序。 PyCharm APIs Spark Connect 使用客户端-服务器架构，将您的应用程序代码与 Spark 驱动程序进程分离。当 Spark 操作在 EMR Serverless 计算上运行时，你可以在本地 IDE 中开发和调试 PySpark 代码。Spark Connect 提供以下好处：

Spark Connect 会话是您的本地 PySpark 客户端与在 Amazon EMR Serverless 上运行的 Spark 驱动程序之间的托管连接。当您启动会话时，EMR Serverless 会代表您配置 Spark 驱动程序和执行器。您的本地客户端向驱动程序发送 DataFrame和 SQL 操作，驱动程序会远程运行它们。会话将一直持续到您终止会话或达到空闲超时为止，因此您无需重新启动 Spark 即可交互式运行多个查询。每个会话都有自己的终端节点 URL 和用于连接的身份验证令牌。

所需的权限

除了访问亚马逊 EMR Serverless 所需的权限外，还要向你的 IAM 角色添加以下权限，以访问 Spark Connect 终端节点和管理 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"
                }
            }
        }
    ]
}

使用交互式会话

要创建支持 Spark Connect 的应用程序并连接到该应用程序，请按照以下步骤操作。

以下是启动会话并将请求直接发送到会话端点的 Python 脚本示例：

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

要访问实时的 Spark 用户界面或 Spark 历史服务器以进行会话，请使用 GetResourceDashboard API。

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

当会话处于活动状态时，URL 会打开实时 Apache Spark 用户界面，用于实时监控查询、阶段和执行器。会话结束后，Spark 历史服务器仍可通过 Amazon EMR Serverless 控制台进行会话后分析。

注意事项和限制

通过 Spark Connect 运行交互式工作负载时，请考虑以下几点。