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

借助 Amazon EMR on EKS 版本emr-7.14.0及更高版本（或更高版本），您可以从、emr-spark-8.1.0和笔记本电脑等VS Code自管理PySpark客户端，连接到托管的 Spark Connect 终端节点。PyCharm JupyterSpark Connect 使用客户端-服务器架构将您的应用程序代码与 Spark 驱动程序进程分离。您可以在本地 IDE 中开发和调试PySpark代码，同时 Spark 操作通过 EKS 上的 Amazon EMR 在 EKS 集群上运行。Spark Connect 具有以下优点：

通过任何PySpark客户端（包括VS Code、和Jupyter笔记本电脑）在 EKS 上连接到 Amazon EMR PyCharm
远程 DataFrames运行生产级数据时，在 IDE 中设置断点并逐步执行PySpark代码
在您自己的 EKS 集群上运行，完全控制计算、网络和安全配置

Spark Connect 终端节点是托管 Spark Connect 服务器的虚拟集群上的托管终端节点。当您创建 Spark Connect 终端节点时，EKS 上的 Amazon EMR 会在您的 EKS 集群上使用 gRPC 服务器预置一个 Spark 驱动程序。您的本地 PySpark 客户端通过 gRPC 端点向驱动程序发送 DataFrame SQL 操作。要与终端节点进行交互，您可以使用 GetManagedEndpointSessionCredentials API 获取会话证书。每个端点支持多个并发会话。

先决条件

在创建 Spark Connect 终端节点之前，请确保您具备以下条件。

以下要求特定于 Spark Connect 端点：

具有至少一个私有子网的 EKS 集群。为了将 gRPC 流量路由到终端节点，Spark Connect 预置了一个内部网络负载均衡器 (NLB)，这需要您的 VPC 中有一个私有子网。NLB 是内部的，不暴露在公共互联网上。
Amazon 负载均衡器控制器安装在您的 EKS 集群上，这样 EKS 上的 Amazon EMR 就可以预置内部 NLB。有关说明，请参阅安装 Amazon 负载均衡器控制器。

您还需要在 EKS 资源上使用以下标准 Amazon EMR。如果你已经在 EKS 上的 Amazon EMR 上运行任务，那么你可能已经准备好了：

一个 IAM 任务执行角色，有权访问您的数据源，例如 Amazon S3 存储桶和数据目录。有关说明，请参阅创建任务执行角色。
如果您的 EKS 集群使用集群访问管理，则为 EKS 上的 Amazon EMR 所需的访问条目。有关说明，请参阅在 EKS 上设置 Amazon EMR 的集群访问权限。

所需的权限

向您的 IAM 角色添加以下权限，以创建 Spark Connect 终端节点并与之交互：


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "EMRContainersEndpointAccess",
            "Effect": "Allow",
            "Action": [
                "emr-containers:CreateManagedEndpoint",
                "emr-containers:DescribeManagedEndpoint",
                "emr-containers:DeleteManagedEndpoint",
                "emr-containers:ListManagedEndpoints",
                "emr-containers:GetManagedEndpointSessionCredentials"
            ],
            "Resource": [
                "arn:aws:emr-containers:region:account-id:/virtualclusters/virtual-cluster-id",
                "arn:aws:emr-containers:region:account-id:/virtualclusters/virtual-cluster-id/endpoints/*"
            ]
        },
        {
            "Sid": "PassRoleToEKSPodIdentity",
            "Effect": "Allow",
            "Action": "iam:PassRole",
            "Resource": "arn:aws:iam::account-id:role/ExecutionRole",
            "Condition": {
                "StringEquals": {
                    "iam:PassedToService": "pods.eks.amazonaws.com"
                },
                "ArnLike": {
                    "iam:AssociatedResourceARN": [
                        "arn:aws:eks:region:account-id:cluster/eks-cluster-name"
                    ]
                }
            }
        }
    ]
}

创建安全配置

Spark Connect 终端节点需要与您的虚拟集群关联的安全配置。安全配置定义端点的身份验证和授权设置，并指定运行 Spark Connect 基础架构的系统命名空间。

注意

每种安全配置与虚拟群集都有一对一的关系。您不能在多个虚拟集群中重复使用相同的安全配置。

使用系统命名空间创建安全配置：


aws emr-containers create-security-configuration \
  --name "security-config-name" \
  --security-configuration-data '{
    "authenticationConfiguration": {
      "identityCenterConfiguration": {
        "enableIdentityCenter": false
      }
    }
  }' \
  --container-provider '{
    "type": "EKS",
    "id": "eks-cluster-name",
    "info": {
      "eksInfo": {
        "namespace": "system-namespace"
      }
    }
  }'

安全配置namespace中的是运行 Spark Connect 基础架构组件的系统命名空间。这与创建虚拟集群时指定的用户命名空间是分开的。

创建启用 Spark Connect 的虚拟集群

创建与安全配置关联的虚拟群集。必须设置sessionEnabled为true并提供上一步securityConfigurationId中的内容。


aws emr-containers create-virtual-cluster \
  --name "virtual-cluster-name" \
  --container-provider '{
    "type": "EKS",
    "id": "eks-cluster-name",
    "info": {
      "eksInfo": {
        "namespace": "user-namespace"
      }
    }
  }' \
  --security-configuration-id SECURITY_CONFIGURATION_ID \
  --session-enabled true

重要

--security-configuration-id— 将此虚拟群集与上一步中创建的安全配置相关联。这是 Spark Connect 端点所必需的。

--session-enabled— 在虚拟集群上启用 Spark Connect 终端节点支持。如果没有此标志，则无法在此虚拟集群上创建 Spark Connect 终端节点。

创建 Spark Connect 端点

创建安全配置后，在虚拟集群上创建 Spark Connect 托管终端节点。

注意

EKS 集群上的第一个 Spark Connect 终端节点ACTIVE比后续的终端节点需要更长的时间才能建立。对于 EKS 集群上的第一个终端节点，EKS 上的 Amazon EMR 预置了用于连接的共享网络组件——内部网络负载均衡器 (NLB)、VPC 接口终端节点 (Amazon PrivateLink) 以及将 gRPC 流量路由到终端节点的一次性 Envoy 路由器部署（在spark-connect-router命名空间中），这可能需要几分钟。这些组件仅在每个 EKS 集群中创建一次，并可由该集群上所有后续终端节点重复使用，因此后续终端节点的启动速度更快。


aws emr-containers create-managed-endpoint \
  --virtual-cluster-id VIRTUAL_CLUSTER_ID \
  --name "spark-connect-endpoint" \
  --type "SPARK_CONNECT" \
  --release-label "emr-7.14.0-latest" \
  --execution-role-arn "arn:aws:iam::account-id:role/ExecutionRole" \
  --session-idle-timeout-in-minutes 60

注意

默认情况下，Spark Connect 托管的端点以 2 个执行器开始。如果您不提供配置替代，则终端节点将使用此默认值。要使用不同数量的执行器运行，请在配置覆盖spark.executor.instances中进行设置，如以下示例所示。

以下示例包括使用动态分配覆盖的 Spark 配置以及用于将 Spark 日志导出到 Amazon S3 的监控配置：


aws emr-containers create-managed-endpoint \
  --virtual-cluster-id VIRTUAL_CLUSTER_ID \
  --name "spark-connect-endpoint" \
  --type "SPARK_CONNECT" \
  --release-label "emr-7.14.0-latest" \
  --execution-role-arn "arn:aws:iam::account-id:role/ExecutionRole" \
  --session-idle-timeout-in-minutes 60 \
  --configuration-overrides '{
    "applicationConfiguration": [
      {
        "classification": "spark-defaults",
        "properties": {
          "spark.driver.memory": "4g",
          "spark.executor.memory": "4g",
          "spark.executor.cores": "2",
          "spark.executor.instances": "3",
          "spark.dynamicAllocation.enabled": "true",
          "spark.dynamicAllocation.minExecutors": "3",
          "spark.dynamicAllocation.maxExecutors": "5"
        }
      }
    ],
    "monitoringConfiguration": {
      "s3MonitoringConfiguration": {
        "logUri": "s3://your-bucket/spark-connect-logs/"
      },
      "persistentAppUI": "ENABLED"
    }
  }'

监控端点状态：


aws emr-containers describe-managed-endpoint \
  --virtual-cluster-id VIRTUAL_CLUSTER_ID \
  --id ENDPOINT_ID

等到端点状态变为连接ACTIVE之前。

连接到 Spark Connect 端点

端点处于活动状态后，获取会话凭据并从 PySpark客户端进行连接。

连接到 Spark Connect 端点

从端点描述中获取身份验证代理 URL：


aws emr-containers describe-managed-endpoint \
  --virtual-cluster-id VIRTUAL_CLUSTER_ID \
  --id ENDPOINT_ID

当端点为时，响应包含该authProxyUrl字段ACTIVE。

获取终端节点的会话令牌：


aws emr-containers get-managed-endpoint-session-credentials \
  --virtual-cluster-identifier VIRTUAL_CLUSTER_ID \
  --endpoint-identifier ENDPOINT_ID \
  --execution-role-arn "arn:aws:iam::account-id:role/ExecutionRole" \
  --credential-type "TOKEN"

响应包含会话令牌：


{
    "id": "SESSION_ID",
    "credentials": {
        "token": "SESSION_TOKEN"
    },
    "endpointCredentials": {
        "token": "ENDPOINT_CREDENTIALS_TOKEN"
    },
    "expiresAt": "EXPIRY_TIME"
}

连接到身份验证代理 URL 时，使用该credentials.token值作为x-aws-proxy-auth参数。

在您的端点上安装与 Spark 版本匹配的 PySpark 客户端（Spark 3.5.8 适用于emr-7.14.0，Spark 4.1.1 适用于emr-spark-8.1.0）：
```
# For emr-7.14.0
pip install pyspark[connect]==3.5.8

# For emr-spark-8.1.0
pip install pyspark[connect]==4.1.1

pip install boto3
```

PySpark 使用身份验证代理 URL 和会话令牌进行连接：


from pyspark.sql import SparkSession

# Use the authProxyUrl from DescribeManagedEndpoint
# and the credentials.token from GetManagedEndpointSessionCredentials
connect_url = f"{auth_proxy_url}/;use_ssl=true;x-aws-proxy-auth={token}"

spark = SparkSession.builder.remote(connect_url).getOrCreate()
print(f"Connected. Spark version: {spark.version}")

# Run queries
spark.sql("SELECT 1+1 AS result").show()

# When finished, disconnect the client
spark.stop()

以下 Python 脚本获取身份验证代理 URL 和会话令牌，然后连接到 Spark Connect 端点：


import boto3
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

REGION = 'REGION'
VIRTUAL_CLUSTER_ID = 'VIRTUAL_CLUSTER_ID'
ENDPOINT_ID = 'ENDPOINT_ID'
EXECUTION_ROLE = 'arn:aws:iam::account-id:role/ExecutionRole'

client = boto3.client('emr-containers', region_name=REGION)

# Get the auth proxy URL from DescribeManagedEndpoint
endpoint_response = client.describe_managed_endpoint(
    virtualClusterId=VIRTUAL_CLUSTER_ID,
    id=ENDPOINT_ID
)
auth_proxy_url = endpoint_response['endpoint']['authProxyUrl']

# Get session token
creds_response = client.get_managed_endpoint_session_credentials(
    virtualClusterIdentifier=VIRTUAL_CLUSTER_ID,
    endpointIdentifier=ENDPOINT_ID,
    executionRoleArn=EXECUTION_ROLE,
    credentialType='TOKEN'
)
token = creds_response['credentials']['token']

# Connect via Spark Connect using auth proxy URL and credentials token
connect_url = f"{auth_proxy_url}/;use_ssl=true;x-aws-proxy-auth={token}"
spark = SparkSession.builder.remote(connect_url).getOrCreate()
print(f"Connected. Spark version: {spark.version}")

# Run DataFrame operations
df = spark.range(100).withColumn("squared", col("id") * col("id"))
df.show(10)
print(f"Count: {df.count()}")

spark.stop()

注意事项和限制

在 EKS 上通过 Amazon EMR 上的 Spark Connect 运行交互式工作负载时，请考虑以下事项。

亚马逊 EMR 在 EKS 版本emr-7.14.0及更高版本或emr-spark-8.1.0更高版本中支持 Spark Connect。
Spark Connect 支持 DataFrame 和 SQL API PySpark。Spark Connect 不支持 RDD-based API。
会话令牌有时间限制。当令牌过期时，gRPC 调用会因身份验证错误而失败。调GetManagedEndpointSessionCredentials用获取新令牌并SparkSession使用更新后的令牌创建新令牌。
每种安全配置与虚拟群集都有一对一的关系。您不能在多个虚拟集群之间共享安全配置。
必须先使用安全配置删除所有终端节点，然后才能删除安全配置。
本地安装的 PySpark 版本必须与终端节点上的 Apache Spark 版本相匹配（Spark 3.5.8emr-7.14.0，Spark 4.1.1 适用于emr-spark-8.1.0）。版本不匹配会导致连接错误或意外行为。
Spark Connect 端点类型为SPARK_CONNECT。这与 Livy 交互式端点（类型JUPYTER_ENTERPRISE_GATEWAY）不同。
该sessionIdleTimeoutInMinutes参数控制空闲会话在自动终止之前持续多长时间。默认值为 60 分钟。
Spark Connect 端点不支持可信身份传播。
Spark Connect 端点尚不支持 Lake Formation 精细访问控制 (FGAC)。要强制执行访问控制，请使用与终端节点关联的 IAM 执行角色。
Spark Connect 端点使用网络负载均衡器 (NLB) 来路由 gRPC 流量。NLB 是在创建第一个 Spark Connect 终端节点时创建的，并且仅在删除最后一个启用会话的虚拟集群时删除。EKS 上的 Amazon EMR 还运行一个 Envoy 路由器（在spark-connect-router命名空间中，每个 EKS 集群一个），用于将 gRPC 流量路由到终端节点；它使用第一个 Spark Connect 终端节点创建，并在删除最后一个启用会话的虚拟集群时终止。除了 Spark 驱动程序和执行器在会话期间消耗的 EKS 计算资源外，您还要承担 NLB 成本和 Envoy 路由器消耗的 EKS 计算。
Python UDF (@udf,spark.udf.register) 要求本地 Python 次要版本与PYTHON_VERSION_MISMATCH远程工作人员版本相匹配，否则它们会失败。 Built-in SQL 函数和 DataFrame 操作不需要 Python 版本匹配。

Javascript 在您的浏览器中被禁用或不可用。

要使用 Amazon Web Services 文档，必须启用 Javascript。请参阅浏览器的帮助页面以了解相关说明。

卸载

运行交互式工作负载