翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

Spark Connect を使用して Amazon EMR on EKS でインタラクティブセッションを実行する

Amazon EMR on EKS リリース emr-7.14.0 以降 (emr-spark-8.1.0または 以降) では、VS Code、、PyCharmJupyterノートブックなどのセルフマネージドPySparkクライアントからマネージド Spark Connect エンドポイントに接続できます。Spark Connect は、アプリケーションコードを Spark ドライバープロセスから切り離すクライアント/サーバーアーキテクチャを使用します。Spark オペレーションが Amazon EMR on EKS を介して EKS クラスターで実行される間、ローカル IDE でPySparkコードを開発およびデバッグします。Spark Connect には次の利点があります。

Spark Connect エンドポイントは、Spark Connect サーバーをホストする仮想クラスター上のマネージドエンドポイントです。Spark Connect エンドポイントを作成すると、Amazon EMR on EKS は EKS クラスターに gRPC サーバーを使用して Spark ドライバーをプロビジョニングします。ローカル PySpark クライアントは、gRPC エンドポイントを介して DataFrame および SQL オペレーションをドライバーに送信します。エンドポイントを操作するには、 GetManagedEndpointSessionCredentials API を使用してセッション認証情報を取得します。各エンドポイントは、複数の同時セッションをサポートします。

前提条件

Spark Connect エンドポイントを作成する前に、以下があることを確認してください。

以下の要件は Spark Connect エンドポイントに固有のものです。

また、以下の標準 Amazon EMR on EKS リソースも必要です。Amazon EMR on EKS でジョブをすでに実行している場合は、次のような設定になっている可能性があります。

必要なアクセス許可

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

Spark Connect エンドポイントを作成する

セキュリティ設定を作成したら、仮想クラスターに Spark Connect マネージドエンドポイントを作成します。

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 設定の上書きと、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 クライアントから接続します。

次の 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()

考慮事項と制限事項

Amazon EMR on EKS で Spark Connect を使用してインタラクティブワークロードを実行する場合は、次の点を考慮してください。