本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 通过 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 负载均衡器控制器](https://docs.amazonaws.cn/eks/latest/userguide/aws-load-balancer-controller.html)。 您还需要在 EKS 资源上使用以下标准 Amazon EMR。如果你已经在 EKS 上的 Amazon EMR 上运行任务，那么你可能已经准备好了： + 一个 IAM 任务执行角色，有权访问您的数据源，例如 Amazon S3 存储桶和数据目录。有关说明，请参阅[创建任务执行角色](https://docs.amazonaws.cn/emr/latest/EMR-on-EKS-DevelopmentGuide/creating-job-execution-role.html)。 + 如果您的 EKS 集群使用集群访问管理，则为 EKS 上的 Amazon EMR 所需的访问条目。有关说明，请参阅[在 EKS ](https://docs.amazonaws.cn/emr/latest/EMR-on-EKS-DevelopmentGuide/setting-up-cluster-access.html) 上设置 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 端点** 1. 从端点描述中获取身份验证代理 URL： ``` aws emr-containers describe-managed-endpoint \ --virtual-cluster-id {{VIRTUAL_CLUSTER_ID}} \ --id {{ENDPOINT_ID}} ``` 当端点为时，响应包含该`authProxyUrl`字段`ACTIVE`。 1. 获取终端节点的会话令牌： ``` 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`参数。 1. 在您的端点上安装与 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 ``` 1. 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.8`emr-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 版本匹配。