通过 Apache Livy 端点在 EMR Serverless 中运行交互式工作负载

在 Amazon EMR 6.14.0 及更高版本中，您可以创建并启用 Apache Livy 端点，同时创建 EMR Serverless 应用程序，以通过自托管笔记本或自定义客户端运行交互式工作负载。Apache Livy 端点具有以下优势：

先决条件

要在 EMR Serverless 中使用 Apache Livy 端点，必须满足以下要求：

所需的权限

除了访问 EMR Serverless 所需的权限外，您还必须向 IAM 角色添加以下权限，才能访问 Apache Livy 端点和运行应用程序：

{
    "Version": "2012-10-17",
    "Statement": [{
            "Sid": "EMRServerlessInteractiveAccess",
            "Effect": "Allow",
            "Action": "emr-serverless:AccessLivyEndpoints",
            "Resource": "arn:aws:emr-serverless:<AWS_REGION>:account:/applications/*"
        },
        {
            "Sid": "EMRServerlessRuntimeRoleAccess",
            "Effect": "Allow",
            "Action": "iam:PassRole",
            "Resource": "execution-role-ARN",
            "Condition": {
                "StringLike": {
                    "iam:PassedToService": "emr-serverless.amazonaws.com"
                }
            }
        },
        {
            "Sid": "EMRServerlessDashboardAccess",
            "Effect": "Allow",
            "Action": "emr-serverless:GetDashboardForJobRun",
            "Resource": "arn:aws:emr-serverless:<AWS_REGION>:account:/applications/*"
        }
    ]
}

开始使用

要创建并运行支持 Apache Livy 的应用程序，请按照以下步骤操作。

端点准备就绪后，就可以根据用例提交工作负载。您必须使用 SIGv4 协议签署对端点的每个请求，并传入授权标头。您可以使用以下方法运行工作负载：

HTTP 客户端

要创建 Apache Livy 会话，您必须在请求正文的 conf 参数中提交 emr-serverless.session.executionRoleArn。下面是一个 POST /sessions 请求示例。

{    
    "kind": "pyspark",
    "heartbeatTimeoutInSecond": 60, 
    "conf": {
        "emr-serverless.session.executionRoleArn": "<executionRoleArn>"
    }
}

下表列出了所有可用的 Apache Livy API 操作。

向 Apache Livy 端点发送请求

您也可以从 HTTP 客户端直接向 Apache Livy 端点发送请求。这样，您就可以在笔记本之外远程运行用例的代码。

在开始向端点发送请求之前，确保已安装以下库：

pip3 install botocore awscrt requests

下面是一个直接向端点发送 HTTP 请求的 Python 脚本示例：

from botocore import crt
import requests 
from botocore.awsrequest import AWSRequest
from botocore.credentials import Credentials
import botocore.session
import json, pprint, textwrap

endpoint = 'https://<application_id>.livy.emr-serverless-services-<AWS_REGION>.amazonaws.com'
headers = {'Content-Type': 'application/json'}

session = botocore.session.Session()
signer = crt.auth.CrtS3SigV4Auth(session.get_credentials(), 'emr-serverless', '<AWS_REGION>')


### Create session request

data = {'kind': 'pyspark', 'heartbeatTimeoutInSecond': 60, 'conf': { 'emr-serverless.session.executionRoleArn': 'arn:aws:iam::123456789012:role/role1'}}

request = AWSRequest(method='POST', url=endpoint + "/sessions", data=json.dumps(data), headers=headers)

request.context["payload_signing_enabled"] = False

signer.add_auth(request)

prepped = request.prepare()

r = requests.post(prepped.url, headers=prepped.headers, data=json.dumps(data))

pprint.pprint(r.json())


### List Sessions Request

request = AWSRequest(method='GET', url=endpoint + "/sessions", headers=headers)

request.context["payload_signing_enabled"] = False

signer.add_auth(request)

prepped = request.prepare()

r2 = requests.get(prepped.url, headers=prepped.headers)
pprint.pprint(r2.json())


### Get session state

session_url = endpoint + r.headers['location']

request = AWSRequest(method='GET', url=session_url, headers=headers)

request.context["payload_signing_enabled"] = False

signer.add_auth(request)

prepped = request.prepare()

r3 = requests.get(prepped.url, headers=prepped.headers)

pprint.pprint(r3.json())


### Submit Statement

data = {
      'code': "1 + 1"
}

statements_url = endpoint + r.headers['location'] + "/statements"

request = AWSRequest(method='POST', url=statements_url, data=json.dumps(data), headers=headers)

request.context["payload_signing_enabled"] = False

signer.add_auth(request)

prepped = request.prepare()

r4 = requests.post(prepped.url, headers=prepped.headers, data=json.dumps(data))

pprint.pprint(r4.json())

### Check statements results

specific_statement_url = endpoint + r4.headers['location']

request = AWSRequest(method='GET', url=specific_statement_url, headers=headers)

request.context["payload_signing_enabled"] = False

signer.add_auth(request)

prepped = request.prepare()

r5 = requests.get(prepped.url, headers=prepped.headers)

pprint.pprint(r5.json())


### Delete session


session_url = endpoint + r.headers['location']

request = AWSRequest(method='DELETE', url=session_url, headers=headers)

request.context["payload_signing_enabled"] = False

signer.add_auth(request)

prepped = request.prepare()

r6 = requests.delete(prepped.url, headers=prepped.headers)

pprint.pprint(r6.json())

Sparkmagic 内核

在安装 Sparkmagic 之前，确保您已在要安装 Sparkmagic 的实例中配置了 AWS 凭证

在内部，每条指令都通过配置的 Apache Livy 端点 URL 调用每个 Apache Livy API 操作。然后，您可以根据自己的用例编写指令。

注意事项

通过 Apache Livy 端点运行交互式工作负载时，请考虑以下注意事项。