This topic describes how to submit Spark jobs, query the status and logs of Spark jobs, terminate Spark jobs, and query Spark historical jobs by using the Python SDK.
Prerequisites
-
A Python environment is installed and the Python version is 3.7 or later.
-
A Job resource group is created in an AnalyticDB for MySQL cluster. For more information, see Create a Job resource group.
-
The AnalyticDB for MySQL cluster is of the Enterprise Edition, Basic Edition, or Lakehouse Edition.
-
The Python SDK is installed. For more information, see AnalyticDB MySQL SDK for Python.
-
The environment variables
ALIBABA_CLOUD_ACCESS_KEY_IDandALIBABA_CLOUD_ACCESS_KEY_SECRETare configured. For more information, see Configure environment variables in Linux, macOS, and Windows. -
The storage path for Spark logs is configured.
NoteYou can configure the storage path for Spark logs by using one of the following methods:
-
On the AnalyticDB for MySQL console, go to the Spark Jar Development page and click Log Configuration in the upper-right corner to set the storage path for Spark logs.
-
Use the configuration item
spark.app.log.rootPathto specify an OSS path for storing Spark job execution logs.
-
Examples
The following sample code shows how to submit Spark jobs, query the status and logs of Spark jobs, terminate Spark jobs, and query Spark historical jobs.
from alibabacloud_adb20211201.models import SubmitSparkAppRequest, SubmitSparkAppResponse, GetSparkAppStateRequest, \
GetSparkAppStateResponse, GetSparkAppLogResponse, GetSparkAppLogRequest, KillSparkAppRequest, \
KillSparkAppResponse, ListSparkAppsRequest, ListSparkAppsResponse
from alibabacloud_tea_openapi.models import Config
from alibabacloud_adb20211201.client import Client
import os
def submit_spark_sql(client: Client, cluster_id, rg_name, sql):
"""
Submit a Spark SQL job
:param client: Alibaba Cloud client
:param cluster_id: Cluster ID
:param rg_name: Resource group name
:param sql: SQL
:return: Spark job ID
:rtype: basestring
:exception ClientException
"""
# Initialize the request
request = SubmitSparkAppRequest(
dbcluster_id=cluster_id,
resource_group_name=rg_name,
data=sql,
app_type="SQL",
agent_source="Python SDK",
agent_version="1.0.0"
)
# Submit the SQL job and obtain the result
response: SubmitSparkAppResponse = client.submit_spark_app(request)
# Obtain the Spark job ID
print(response)
return response.body.data.app_id
def submit_spark_jar(client: Client, cluster_id: str, rg_name: str, json_conf: str):
"""
Submit a Spark job
:param client: Alibaba Cloud client
:param cluster_id: Cluster ID
:param rg_name: Resource group name
:param json_conf: JSON configuration
:return: Spark job ID
:rtype: basestring
:exception ClientException
"""
# Initialize the request
request = SubmitSparkAppRequest(
dbcluster_id=cluster_id,
resource_group_name=rg_name,
data=json_conf,
app_type="BATCH",
agent_source="Python SDK",
agent_version="1.0.0"
)
# Submit the SQL job and obtain the result
response: SubmitSparkAppResponse = client.submit_spark_app(request)
# Obtain the Spark job ID
print(response)
return response.body.data.app_id
def get_status(client: Client, app_id):
"""
Query the status of a Spark job
:param client: Alibaba Cloud client
:param app_id: Spark job ID
:return: Status of the Spark job
:rtype: basestring
:exception ClientException
"""
# Initialize the request
print(app_id)
request = GetSparkAppStateRequest(app_id=app_id)
# Obtain the status of the Spark job
response: GetSparkAppStateResponse = client.get_spark_app_state(request)
print(response)
return response.body.data.state
def get_log(client: Client, app_id):
"""
Query the logs of a Spark job
:param client: Alibaba Cloud client
:param app_id: Spark job ID
:return: Logs of the Spark job
:rtype: basestring
:exception ClientException
"""
# Initialize the request
request = GetSparkAppLogRequest(app_id=app_id)
# Obtain the logs of the Spark job
response: GetSparkAppLogResponse = client.get_spark_app_log(request)
print(response)
return response.body.data.log_content
def kill_app(client: Client, app_id):
"""
Terminate a Spark job
:param client: Alibaba Cloud client
:param app_id: Spark job ID
:return: Status of the Spark job
:exception ClientException
"""
# Initialize the request
request = KillSparkAppRequest(app_id=app_id)
# Obtain the status of the Spark job
response: KillSparkAppResponse = client.kill_spark_app(request)
print(response)
return response.body.data.state
def list_apps(client: Client, cluster_id: str, page_number: int, page_size: int):
"""
Query Spark historical jobs
:param client: Alibaba Cloud client
:param cluster_id: Cluster ID
:param page_number: Page number, which must be a positive integer. Default value: 1
:param page_size: Number of entries per page
:return: Spark job details
:exception ClientException
"""
# Initialize the request
request = ListSparkAppsRequest(
dbcluster_id=cluster_id,
page_number=page_number,
page_size=page_size
)
# Obtain Spark job details
response: ListSparkAppsResponse = client.list_spark_apps(request)
print("Total App Number:", response.body.data.page_number)
for app_info in response.body.data.app_info_list:
print(app_info.app_id)
print(app_info.state)
print(app_info.detail)
if __name__ == '__main__':
# client config
config = Config(
# Obtain the AccessKey ID from the environment variable ALIBABA_CLOUD_ACCESS_KEY_ID
access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
# Obtain the AccessKey Secret from the environment variable ALIBABA_CLOUD_ACCESS_KEY_SECRET
access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'],
# Endpoint. cn-hangzhou is the region ID of the cluster.
endpoint="adb.cn-hangzhou.aliyuncs.com"
)
# new client
adb_client = Client(config)
sql_str = """
-- Here is just an example of SparkSQL. Modify the content and run your spark program.
set spark.driver.resourceSpec=medium;
set spark.executor.instances=2;
set spark.executor.resourceSpec=medium;
set spark.app.name=Spark SQL Test;
-- Here are your sql statements
show databases;
"""
json_str = """
{
"comments": [
"-- Here is just an example of SparkPi. Modify the content and run your spark program."
],
"args": [
"1000"
],
"file": "local:///tmp/spark-examples.jar",
"name": "SparkPi",
"className": "org.apache.spark.examples.SparkPi",
"conf": {
"spark.driver.resourceSpec": "medium",
"spark.executor.instances": 2,
"spark.executor.resourceSpec": "medium"
}
}
"""
"""
Submit a Spark SQL job
cluster_id: Cluster ID
rg_name: Resource group name
"""
sql_app_id = submit_spark_sql(client=adb_client, cluster_id="amv-bp1wo70f0k3c****", rg_name="test", sql=sql_str)
print(sql_app_id)
"""
Submit a Spark job
cluster_id: Cluster ID
rg_name: Resource group name
"""
json_app_id = submit_spark_jar(client=adb_client, cluster_id="amv-bp1wo70f0k3c****",
rg_name="test", json_conf=json_str)
print(json_app_id)
# Query the status of the Spark job
get_status(client=adb_client, app_id=sql_app_id)
get_status(client=adb_client, app_id=json_app_id)
"""
Query Spark historical jobs
cluster_id: Cluster ID
page_number: Page number, which must be a positive integer. Default value: 1
page_size: Number of entries per page
"""
list_apps(client=adb_client, cluster_id="amv-bp1wo70f0k3c****", page_size=10, page_number=1)
# Terminate the Spark job
kill_app(client=adb_client, app_id=json_app_id)