Bu konu, Python SDK kullanarak Spark işlerinin nasıl gönderileceğini, Spark işlerinin durumunun ve günlüklerinin nasıl sorgulanacağını, Spark işlerinin nasıl sonlandırılacağını ve Spark geçmiş işlerinin nasıl sorgulanacağını açıklar.
Ön Koşullar
Bir Python ortamı yüklüdür ve Python sürümü 3,7 veya üstüdür.
Bir AnalyticDB for MySQL kümesinde bir Job kaynak grubu oluşturulmuştur. Daha fazla bilgi için bkz. Job kaynak grubu oluşturma.
AnalyticDB for MySQL kümesi Enterprise Edition, Basic Edition veya Lakehouse Edition'dır.
Python SDK yüklenmiştir. Daha fazla bilgi için bkz. AnalyticDB MySQL SDK for Python.
ALIBABA_CLOUD_ACCESS_KEY_IDveALIBABA_CLOUD_ACCESS_KEY_SECRETortam değişkenleri yapılandırılmıştır. Daha fazla bilgi için bkz. Linux, macOS ve Windows'ta ortam değişkenlerini yapılandırma.-
Spark günlükleri için depolama yolu yapılandırılmıştır.
NotSpark günlükleri için depolama yolunu aşağıdaki yöntemlerden birini kullanarak yapılandırabilirsiniz:
AnalyticDB for MySQL konsolunda, Spark Jar Development sayfasına gidin ve sağ üst köşede Log Configuration seçeneğine tıklayarak Spark günlükleri için depolama yolunu ayarlayın.
Spark işi yürütme günlüklerini depolamak üzere bir OSS yolu belirtmek için
spark.app.log.rootPathyapılandırma öğesini kullanın.
Örnekler
Aşağıdaki örnek kod, Spark işlerinin nasıl gönderileceğini, Spark işlerinin durumunun ve günlüklerinin nasıl sorgulanacağını, Spark işlerinin nasıl sonlandırılacağını ve Spark geçmiş işlerinin nasıl sorgulanacağını gösterir.
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)