通过 Spark Connect 与 Amazon EMR 无服务器运行互动会话

在亚马逊 EMR 版本emr-7.13.0及更高版本中，您可以使用带有 Apache Spark Connect 的 EMR Serverless 会话 API 从 VS Code 和 Jupyter 笔记本电脑等自管理 PySpark 客户端 PyCharm，连接到亚马逊 EMR Serverless 应用程序。Spark Connect 使用客户端-服务器架构将您的应用程序代码与 Spark 驱动程序进程分离。您可以在本地 IDE 中开发和调试 PySpark 代码，而 Spark 操作在 EMR 无服务器计算上运行。Spark Connect 具有以下优点：

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

所需的权限

除了访问亚马逊 EMR 无服务器所需的权限外，还要向您的 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: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"
                }
            }
        }
    ]
}

使用交互式会话

要创建 Spark Connect-enabled 应用程序并连接到该应用程序，请执行以下步骤。

以下是启动会话并将请求直接发送到会话端点的示例 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 UI 或 Spark History Server，请使用 GetResourceDashboard API。

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

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

注意事项和限制

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