Este tópico descreve como enviar jobs Spark, consultar o status e os logs desses jobs, encerrá-los e consultar jobs históricos do Spark usando o SDK para Python.
Pré-requisitos
Ambiente Python instalado com versão 3,7 ou superior.
Grupo de recursos de Job criado em um cluster do AnalyticDB for MySQL. Para mais informações, consulte Create a Job resource group.
Cluster do AnalyticDB for MySQL nas edições Enterprise, Basic ou Lakehouse.
SDK para Python instalado. Para mais informações, consulte AnalyticDB MySQL SDK for Python.
Variáveis de ambiente
ALIBABA_CLOUD_ACCESS_KEY_IDeALIBABA_CLOUD_ACCESS_KEY_SECRETconfiguradas. Para mais informações, consulte Configure environment variables in Linux, macOS, and Windows.-
Caminho de armazenamento dos logs do Spark configurado.
NotaConfigure o caminho de armazenamento dos logs do Spark usando um dos métodos a seguir:
No console do AnalyticDB for MySQL, acesse a página Spark Jar Development e clique em Log Configuration no canto superior direito para definir o caminho de armazenamento dos logs do Spark.
Use o item de configuração
spark.app.log.rootPathpara especificar um caminho do OSS para armazenar os logs de execução dos jobs Spark.
Exemplos
O código de exemplo a seguir mostra como enviar jobs Spark, consultar o status e os logs desses jobs, encerrá-los e consultar jobs históricos do Spark.
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)