Spark Connect 지원

Spark Connect는 Spark 클러스터의 드라이버 프로세스에서 애플리케이션 클라이언트를 분리하는 Apache Spark용 클라이언트-서버 아키텍처입니다. 이를 통해 지원되는 클라이언트에서 Spark에 대한 원격 연결을 허용합니다. 또한 Spark Connect는 선호하는 IDE/클라이언트에서 직접 개발하는 동안 대화형 디버깅을 활성화합니다.

Apache Spark 버전 3.5 릴리스 버전부터 Athena는 GetSessionEndpoint API를 사용하여 액세스할 수 있는 AWS 엔드포인트로 Spark Connect를 지원합니다.

API/CLI 예제(GetSessionEndpoint)

GetSessionEndpoint API를 사용하여 대화형 세션의 Spark Connect 엔드포인트를 가져올 수 있습니다.

aws athena get-session-endpoint \
  --region "REGION" \
  --session-id "SESSION_ID"

이 API는 해당 세션에 대한 Spark Connect 엔드포인트 URL을 반환합니다.

{
  "EndpointUrl": "ENDPOINT_URL",
  "AuthToken": "AUTH_TOKEN",
  "AuthTokenExpirationTime": "AUTH_TOKEN_EXPIRY_TIME"
}

자체 관리형 클라이언트에서 연결

자체 관리형 클라이언트에서 Athena Spark 대화형 세션에 연결할 수 있습니다.

필수 조건

Spark 3.5.6용 pyspark 연결 클라이언트와 Python용 AWS SDK를 설치합니다.

pip install --user pyspark[connect]==3.5.6
pip install --user boto3

다음은 요청을 세션 엔드포인트로 직접 전송하기 위한 Python 스크립트 샘플입니다.

import boto3
import time
from pyspark.sql import SparkSession

client = boto3.client('athena', region_name='<REGION>')

# start the session
response = client.start_session(
    WorkGroup='<WORKGROUP_NAME>',
    EngineConfiguration={}
)

# wait for the session endpoint to be ready
time.sleep(5)
response = client.get_session_endpoint(SessionId=session_id)

# construct the authenticated remote url
authtoken=response['AuthToken']
endpoint_url=response['EndpointUrl']
endpoint_url=endpoint_url.replace("https", "sc")+":443/;use_ssl=true;"
url_with_headers = (
    f"{endpoint_url}"
    f"x-aws-proxy-auth={authtoken}"
)

# start the Spark session
start_time = time.time()
spark = SparkSession.builder\
    .remote(url_with_headers)\
    .getOrCreate()
 
spark.version 

#
# Enter your spark code here
#

# stop the Spark session
spark.stop()

다음은 세션의 라이브 Spark UI 또는 Spark 기록 서버에 액세스하기 위한 Python 스크립트 샘플입니다.

Region='<REGION>'
WorkGroupName='<WORKGROUP_NAME>'
SessionId='<SESSION_ID>'
Partition='aws'
Account='<ACCOUNT_NUMBER>'

SessionARN=f"arn:{Partition}:athena:{Region}:{Account}:workgroup/{WorkGroupName}/session/{SessionId}"

# invoke the API to get the live UI/persistence UI for a session
response = client.get_resource_dashboard(
    ResourceARN=SessionARN
)
response['Url']