En este tema se describe cómo enviar trabajos de Spark, consultar su estado y registros, finalizarlos y revisar los trabajos históricos mediante el SDK de Python.
Requisitos previos
Tener instalado un entorno de Python con la versión 3.7 o posterior.
Haber creado un resource group de trabajo en un clúster de AnalyticDB for MySQL. Para obtener más información, consulte Create a Job resource group.
El clúster de AnalyticDB for MySQL debe ser de edición Enterprise, Basic o Lakehouse.
Tener instalado el SDK de Python. Para obtener más información, consulte AnalyticDB MySQL SDK for Python.
Tener configuradas las variables de entorno
ALIBABA_CLOUD_ACCESS_KEY_IDyALIBABA_CLOUD_ACCESS_KEY_SECRET. Para obtener más información, consulte Configure environment variables on Linux, macOS, and Windows.-
Tener configurada la ruta de almacenamiento para los registros de Spark.
NotaConfigure la ruta de almacenamiento de los registros de Spark con alguno de los siguientes métodos:
En la consola de AnalyticDB for MySQL, vaya a la página Spark Jar Development y haga clic en Log Configuration (esquina superior derecha) para establecer la ruta de almacenamiento de los registros de Spark.
Utilice el elemento de configuración
spark.app.log.rootPathpara especificar una ruta de OSS donde almacenar los registros de ejecución de los trabajos de Spark.
Ejemplos
A continuación se muestra un código de ejemplo que ilustra cómo enviar trabajos de Spark, consultar su estado y registros, finalizarlos y revisar los trabajos históricos.
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)