このトピックでは、Python SDK を使用して Spark ジョブを投入する方法、Spark ジョブのステータスとログをクエリする方法、Spark ジョブを終了する方法、および Spark 履歴ジョブをクエリする方法について説明します。
前提条件
-
Python 環境がインストールされており、Python のバージョンが 3.7 以降であること。
-
AnalyticDB for MySQL クラスターにジョブリソースグループが作成されます。詳細については、「ジョブリソースグループの作成」をご参照ください。
-
AnalyticDB for MySQL クラスターは、エンタープライズ版、ベーシック版、レイクハウス版のいずれかです。
-
Python SDK がインストールされていること。詳細については、「AnalyticDB MySQL SDK for Python」をご参照ください。
-
環境変数
ALIBABA_CLOUD_ACCESS_KEY_IDとALIBABA_CLOUD_ACCESS_KEY_SECRETが設定されていること。詳細については、「Linux、macOS、Windows で環境変数を設定する」をご参照ください。 -
Spark ログのストレージパスが設定されていること。
説明次のいずれかの方法で Spark ログのストレージパスを設定できます：
-
AnalyticDB for MySQL コンソールの [Spark Jar 開発] ページで、右上隅にある [ログ設定] をクリックして Spark ログのストレージパスを設定します。
-
設定項目
spark.app.log.rootPathを使用して、Spark ジョブ実行ログを格納する OSS パスを指定します。
-
例
次のサンプルコードは、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):
"""
Spark SQL ジョブを投入します
:param client: Alibaba Cloud クライアント
:param cluster_id: クラスター ID
:param rg_name: リソースグループ名
:param sql: SQL
:return: Spark ジョブ ID
:rtype: basestring
:exception ClientException
"""
# リクエストを初期化します
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"
)
# SQL ジョブを投入して結果を取得します
response: SubmitSparkAppResponse = client.submit_spark_app(request)
# Spark ジョブ ID を取得します
print(response)
return response.body.data.app_id
def submit_spark_jar(client: Client, cluster_id: str, rg_name: str, json_conf: str):
"""
Spark ジョブを投入します
:param client: Alibaba Cloud クライアント
:param cluster_id: クラスター ID
:param rg_name: リソースグループ名
:param json_conf: JSON 設定
:return: Spark ジョブ ID
:rtype: basestring
:exception ClientException
"""
# リクエストを初期化します
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"
)
# Spark ジョブを投入して結果を取得します
response: SubmitSparkAppResponse = client.submit_spark_app(request)
# Spark ジョブ ID を取得します
print(response)
return response.body.data.app_id
def get_status(client: Client, app_id):
"""
Spark ジョブのステータスをクエリします
:param client: Alibaba Cloud クライアント
:param app_id: Spark ジョブ ID
:return: Spark ジョブのステータス
:rtype: basestring
:exception ClientException
"""
# リクエストを初期化します
print(app_id)
request = GetSparkAppStateRequest(app_id=app_id)
# Spark ジョブのステータスを取得します
response: GetSparkAppStateResponse = client.get_spark_app_state(request)
print(response)
return response.body.data.state
def get_log(client: Client, app_id):
"""
Spark ジョブのログをクエリします
:param client: Alibaba Cloud クライアント
:param app_id: Spark ジョブ ID
:return: Spark ジョブのログ
:rtype: basestring
:exception ClientException
"""
# リクエストを初期化します
request = GetSparkAppLogRequest(app_id=app_id)
# Spark ジョブのログを取得します
response: GetSparkAppLogResponse = client.get_spark_app_log(request)
print(response)
return response.body.data.log_content
def kill_app(client: Client, app_id):
"""
Spark ジョブを終了します
:param client: Alibaba Cloud クライアント
:param app_id: Spark ジョブ ID
:return: Spark ジョブのステータス
:exception ClientException
"""
# リクエストを初期化します
request = KillSparkAppRequest(app_id=app_id)
# Spark ジョブを終了します
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):
"""
Spark 履歴ジョブをクエリします
:param client: Alibaba Cloud クライアント
:param cluster_id: クラスター ID
:param page_number: ページ番号。正の整数である必要があります。デフォルト値： 1
:param page_size: 1 ページあたりのエントリ数
:return: Spark ジョブの詳細
:exception ClientException
"""
# リクエストを初期化します
request = ListSparkAppsRequest(
dbcluster_id=cluster_id,
page_number=page_number,
page_size=page_size
)
# Spark ジョブの詳細を取得します
response: ListSparkAppsResponse = client.list_spark_apps(request)
print("Page 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__':
# クライアント設定
config = Config(
# 環境変数 ALIBABA_CLOUD_ACCESS_KEY_ID から AccessKey ID を取得します
access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
# 環境変数 ALIBABA_CLOUD_ACCESS_KEY_SECRET から AccessKey secret を取得します
access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'],
# エンドポイント。cn-hangzhou はクラスターのリージョン ID です。
endpoint="adb.cn-hangzhou.aliyuncs.com"
)
# 新規クライアント
adb_client = Client(config)
sql_str = """
-- これは SparkSQL の一例です。内容を修正して、お使いの Spark プログラムを実行してください。
set spark.driver.resourceSpec=medium;
set spark.executor.instances=2;
set spark.executor.resourceSpec=medium;
set spark.app.name=Spark SQL Test;
-- ここに SQL 文を記述します
show databases;
"""
json_str = """
{
"comments": [
"-- これは SparkPi の一例です。内容を修正して、お使いの Spark プログラムを実行してください。"
],
"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"
}
}
"""
"""
Spark SQL ジョブを投入します
cluster_id: クラスター ID
rg_name: リソースグループ名
"""
sql_app_id = submit_spark_sql(client=adb_client, cluster_id="amv-bp1wo70f0k3c****", rg_name="test", sql=sql_str)
print(sql_app_id)
"""
Spark ジョブを投入します
cluster_id: クラスター ID
rg_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)
# Spark ジョブのステータスをクエリします
get_status(client=adb_client, app_id=sql_app_id)
get_status(client=adb_client, app_id=json_app_id)
"""
Spark 履歴ジョブをクエリします
cluster_id: クラスター ID
page_number: ページ番号。正の整数である必要があります。デフォルト値： 1
page_size: 1 ページあたりのエントリ数
"""
list_apps(client=adb_client, cluster_id="amv-bp1wo70f0k3c****", page_size=10, page_number=1)
# Spark ジョブを終了します
kill_app(client=adb_client, app_id=json_app_id)