MaxCompute:Use PyODPS in DataWorks

メモリ

メモリ使用量がノードの制限を超えると、ノードは Got killed と報告して終了します。これを回避するには、データを DataWorks 環境にダウンロードしてローカルで処理するのではなく、データ処理タスクを MaxCompute にプッシュして分散実行します。2 つのアプローチの比較については、「概要」の「注意事項」セクションをご参照ください。

データレコード

デフォルトでは、DataWorks の PyODPS ノードで options.tunnel.use_instance_tunnel は False に設定されています。これは、instance.open_reader が Result インターフェイスを使用することを意味します。このインターフェイスは最大 10,000 件のデータレコードを返し、複雑なデータ型のサポートには制限があります。すべてのレコードを読み取るか、Array などの複雑な型を処理するには、InstanceTunnel を有効にしてください。詳細については、「SQL 実行結果の読み取り」をご参照ください。

パッケージ

ほとんどの一般的なデータサイエンスパッケージはプリインストール済みです (「プリインストール済みのパッケージ」をご参照ください)。以下の制限が適用されます。

• atexit はサポートされていません。代わりに try-finally を使用してください。

• matplotlib は利用できないため、DataFrame の plot 関数に影響します。

• DataFrame UDF は Python サンドボックスで実行され、純粋な Python ライブラリと NumPy のみを使用できます。pandas などの他のサードパーティライブラリは UDF 内ではサポートされていません。UDF 以外の操作では、NumPy と pandas を利用できます。

• バイナリコードを含むサードパーティパッケージはサポートされていません。

SQL 実行結果の読み取り

DataWorks ではデフォルトで InstanceTunnel が有効になっていないため、instance.open_reader は 10,000 レコードに制限されており、複雑なデータ型をサポートしていない場合があります。ご利用のプロジェクトでデータ保護が有効になっておらず、すべてのレコードを読み取るか、Array などの複雑な型を使用する必要がある場合は、InstanceTunnel を有効にしてください。

オプション 1：グローバルに有効化

ノード内の後続のすべての open_reader 呼び出しに適用されます。

options.tunnel.use_instance_tunnel = True
options.tunnel.limit_instance_tunnel = False  # 10,000 レコードの制限を解除します。

with instance.open_reader() as reader:
    # InstanceTunnel がアクティブです。reader.count を使用してレコードの総数を取得します。
    pass

オプション 2：リーダーごとに有効化

現在の open_reader 呼び出しにのみ適用されます。

with instance.open_reader(tunnel=True, limit=False) as reader:
    # このリーダーに対して InstanceTunnel がアクティブです。すべてのレコードにアクセスできます。
    pass

詳細については、「SQL ステートメントの実行結果の取得」をご参照ください。

プリインストール済みのパッケージ

DataWorks ノードには、次のパッケージがプリインストールされています。

カスタムパッケージのインストール

必要なパッケージがプリインストールされていない場合は、pyodps-pack を使用してバンドルし、load_resource_package を使用してノードにロードします。

1. パッケージをバンドルします。次の例では、ipaddress パッケージをバンドルします。

   pyodps-pack -o ipaddress-bundle.tar.gz ipaddress

   Python 2 ノードの場合は、--dwpy27 を追加します。

   pyodps-pack --dwpy27 -o ipaddress-bundle.tar.gz ipaddress

   バンドルサイズを小さくするには、DataWorks にすでにプリインストールされているパッケージを除外します。

   pyodps-pack -o bundle.tar.gz --exclude numpy --exclude pandas <your-package>

   ダウンロードされたパッケージの合計サイズは 100 MB を超えることはできません。

2. .tar.gz ファイルを MaxCompute リソースとしてアップロードして送信します。

3. PyODPS ノードで、パッケージをロードしてインポートします。

   load_resource_package("ipaddress-bundle.tar.gz")
   import ipaddress

詳細については、「PyODPS 用のサードパーティパッケージの生成」および「PyODPS ノードでのサードパーティパッケージの参照」をご参照ください。

手順

1. 新しいアカウントにプロジェクトに対する必要な権限を付与します。「付録：別のアカウントへの権限付与」をご参照ください。

2. PyODPS ノードで、新しいアカウント用のエントリオブジェクトを作成します。

   import os
   
   # 認証情報をハードコーディングするのではなく、環境変数に保存します。
   new_odps = o.as_account(
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
   )

3. 現在のユーザーを確認して、アカウントの切り替えが成功したことを確認します。

   print(new_odps.get_project().current_user)

   出力が新しいアカウントの AccessKey ID と一致する場合、切り替えは成功です。

例

この例では、テーブルを作成し、別のアカウントを使用してクエリを実行し、結果を出力します。

1. pyodps_iris テーブルを作成し、サンプルデータをインポートします。手順については、「テーブルの作成とデータのアップロード」をご参照ください。

   CREATE TABLE IF NOT EXISTS pyodps_iris
   (
     sepallength  DOUBLE COMMENT 'sepal length (cm)',
     sepalwidth   DOUBLE COMMENT 'sepal width (cm)',
     petallength  DOUBLE COMMENT 'petal length (cm)',
     petalwidth   DOUBLE COMMENT 'petal width (cm)',
     name         STRING COMMENT 'type'
   );

2. 新しいアカウントにプロジェクトとテーブルに対する権限を付与します。「付録：別のアカウントへの権限付与」をご参照ください。

3. PyODPS 3 ノードを作成し、次のコードを実行します。手順については、「PyODPS 3 タスクの開発」をご参照ください。

   from odps import ODPS
   import os
   
   # 認証情報をハードコーディングするのではなく、環境変数に保存します。
   os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'] = '<your-access-key-id>'
   os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'] = '<your-access-key-secret>'
   
   od = o.as_account(
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
   )
   
   # sepallength > 5 の行をクエリします。
   with od.execute_sql('SELECT * FROM pyodps_iris WHERE sepallength > 5').open_reader() as reader:
       print(reader.raw)
       for record in reader:
           print(record["sepallength"], record["sepalwidth"],
                 record["petallength"], record["petalwidth"], record["name"])
   
   # 現在のユーザーを確認します。
   print(od.get_project().current_user)

4. ノードを実行します。出力は次のようになります。

   Executing user script with PyODPS 0.11.4.post0
   
   "sepallength","sepalwidth","petallength","petalwidth","name"
   5.4,3.9,1.7,0.4,"Iris-setosa"
   ...
   <User 139xxxxxxxxxxxxx>