PyODPS でタスクインスタンスを管理する - MaxCompute - Alibaba Cloud - MaxCompute

SQL タスクなどのタスクは、MaxCompute における計算の基本単位です。このトピックでは、PyODPS におけるタスクインスタンスの基本操作について説明します。

基本操作

タスクが実行されると、MaxCompute インスタンスとしてインスタンス化されます。これはインスタンスとも呼ばれます。インスタンスの一般的な操作を次に示します。

list_instances()：プロジェクト内のすべてのインスタンスを取得します。
exist_instance()：インスタンスが存在するかどうかを確認します。
get_instance()：インスタンスを取得します。
stop_instance()：インスタンスを停止します。実行中のインスタンスのみ停止できます。他の状態のインスタンスを停止しようとすると、エラーが発生します。
説明
Instance オブジェクトの stop メソッドを呼び出すことでも、インスタンスを停止できます。例： o.get_instance('my_instance_id').stop()

例

インスタンス ID を取得する

for instance in o.list_instances():
    print(instance.id)

インスタンスが存在するかどうかを確認する
```
print(o.exist_instance('my_instance_id'))
```

Logview アドレスの取得

SQL タスクなどのタスクの場合、Instance オブジェクトの get_logview_address メソッドを呼び出して、Logview アドレスを取得できます。

# 既存のインスタンスオブジェクトから Logview アドレスを取得します。
instance = o.run_sql('desc pyodps_iris')
print(instance.get_logview_address())

# インスタンス ID から Logview アドレスを取得します。
instance = o.get_instance('my_instance_id')
print(instance.get_logview_address())

PAI タスクの場合、まずサブタスクを一覧取得し、各サブタスクの Logview アドレスを取得できます。

instance = o.run_xflow('AppendID', 'algo_public', {'inputTableName': 'input_table', 'outputTableName': 'output_table'})
for sub_inst_name, sub_inst in o.get_xflow_sub_instances(instance).items():
    print('%s: %s' % (sub_inst_name, sub_inst.get_logview_address()))

タスクインスタンスのステータス

インスタンスのステータスは、Running、Suspended、またはTerminated のいずれかです。インスタンスのステータスに関連するメソッドは次のとおりです。

status：ステータスを取得するプロパティです。
is_terminated：インスタンスの実行が終了したかどうかを返します。
is_successful：インスタンスが正常に実行されたかどうかを返します。タスクが実行中または失敗した場合は False を返します。

以下にコード例を示します。

インスタンスのステータスを取得する

instance = o.get_instance('my_instance_id')
print(instance.status)
print(instance.status.value)

戻り値の例：

Status.TERMINATED
Terminated

現在のインスタンスの実行が終了したかどうかを確認する
```
instance = o.get_instance('my_instance_id')
from odps.models import Instance
print(instance.status == Instance.Status.TERMINATED)
```
返り値が True の場合、インスタンスの実行が終了したことを示します。

説明

wait_for_completion メソッドを呼び出すと、インスタンスの実行が終了するまでプロセスがブロックされます。wait_for_success メソッドもプロセスをブロックしますが、タスクが失敗した場合は例外が発生します。

サブタスクの操作

インスタンスには、実行時に 1 つ以上のサブタスクを含めることができます。これらのサブタスクはタスクとも呼ばれます。この Task は、MaxCompute の計算の基本単位であるタスクとは異なる点にご注意ください。これらのタスクには、次のメソッドがあります。

get_task_names：すべてのサブタスクの名前を取得します。このメソッドは、名前のリストを返します。
```
instance = o.get_instance('my_instance_id')
instance.get_task_names()
```
戻り値の例：
```
['jdbc_sql_task']
```

get_task_result：指定されたタスクの実行結果を取得します。このメソッドは、指定されたタスクの実行結果を文字列として返します。以下にコード例を示します。

サブタスク名を取得する

instance = o.execute_sql('select*frompyodps_irislimit1')
print(instance.get_task_names())

戻り値の例：

['AnonymousSQLTask']

サブタスクの実行結果を取得する

print(instance.get_task_result('AnonymousSQLTask'))

戻り値の例：

"sepallength","sepalwidth","petallength","petalwidth","name"
4.9,3.0,1.4,0.2,"Iris-setosa"

サブタスクの結果を取得する

print(instance.get_task_results())

戻り値の例：

OrderedDict([('AnonymousSQLTask', '"sepallength","sepalwidth","petallength","petalwidth","name"\n4.9,3.0,1.4,0.2,"Iris-setosa"\n')])

get_task_progress：インスタンスの実行中にタスクの現在の進行状況を取得します。

instance = o.get_instance('20160519101349613gzbzufck2')
while not instance.is_terminated():
    for task_name in instance.get_task_names():
        print(instance.id, instance.get_task_progress(task_name).get_stage_progress_formatted_string())
        time.sleep(10)

戻り値の例：

20160519101349613gzbzufck2 2016-05-19 18:14:03 M1_Stg1_job0:0/1/1[100%]