MaxCompute:DataWorks で PyODPS を使用する

制限事項

• 使用量の制限

  Got killed エラーが発生した場合、そのプロセスはメモリ制限を超えたために終了されています。DataWorks 内の PyODPS ノードでデータをダウンロードして処理することは避けてください。代わりに、データ処理タスクを MaxCompute に送信して分散実行を行ってください。2 つの方法の比較については、「注意事項：PyODPS を実行するために完全なデータをローカルにダウンロードしない」をご参照ください。

• パッケージサポートの制限

  • DataWorks の PyODPS ノードには、matplotlib などの一部のパッケージが含まれていません。これにより、以下の機能が制限される場合があります。

    • DataFrame の plot 関数。

    • DataFrame のユーザー定義関数 (UDF) は、実行のために MaxCompute に送信する必要があります。Python サンドボックスは、サードパーティ製のライブラリとして純粋な Python ライブラリと NumPy のみをサポートしているため、Pandas を直接使用することはできません。

    • UDF を使用せずに DataWorks で実行されるコードでは、プリインストール済みの NumPy および Pandas パッケージを使用できます。バイナリコードを含む他のサードパーティ製パッケージはサポートされていません。

  • DataWorks の PyODPS ノードは、Python の atexit パッケージをサポートしていません。同様の機能を実現するには、try-finally ブロックを使用してください。

• 読み取り可能なデータレコード数の制限

  デフォルトでは、DataWorks の PyODPS ノードに対して options.tunnel.use_instance_tunnel は False に設定されており、読み取りは最大 10,000 レコードに制限されます。より多くのレコードを読み取るには、options.tunnel.use_instance_tunnel を True に設定して、インスタンストンネルをグローバルに有効にする必要があります。

操作手順

1. PyODPS ノードの作成

   DataWorks の データ開発 ページに移動して、PyODPS ノードを作成します。PyODPS ノードには 2 種類あります。

   • PyODPS 2：Python 2 ベース。

   • PyODPS 3：Python 3 ベース。

   使用する Python のバージョンに基づいて PyODPS ノードを作成します。詳細な手順については、「PyODPS 2 タスクの開発」および「PyODPS 3 タスクの開発」をご参照ください。

2. PyODPS タスクのコード開発

   ノードを作成した後、以下のセクションの例に従って、PyODPS の主な機能について学びます。

   • MaxCompute エントリポイント

   • SQL の実行

   • DataFrame

   • スケジューリングパラメーターの取得

   • 実行時パラメーターヒントの設定

   PyODPS の使用方法の詳細については、「基本操作の概要」および「DataFrame (非推奨)」をご参照ください。また、「PyODPS ノードを使用した Jieba 中国語単語分割」の例に従って、簡単なエンドツーエンドの操作を実行することもできます。

3. スケジューリングプロパティを設定し、ノードを保存、コミット、デプロイして、定期的に実行できるようにします。

MaxCompute エントリポイント

# pyodps_iris テーブルが存在するかどうかを確認します。
print(o.exist_table('pyodps_iris'))

SQL の実行

一般的な機能

注意事項：データ形式とレコード制限

DataFrame

• 実行

  DataWorks の DataFrame 操作では、execute や persist などの 即時実行メソッドを明示的に呼び出す必要があります。例えば:

  # 即時実行メソッドを呼び出して各レコードを処理し、iris.sepalwidth が 3 未満の pyodps_iris テーブルからすべてのデータを出力します。
  from odps.df import DataFrame
  iris = DataFrame(o.get_table('pyodps_iris'))
  for record in iris[iris.sepalwidth < 3].execute():  
      print(record)

• 詳細出力

  デフォルトでは、DataWorks で options.verbose オプションが有効になっており、ノード実行中に Logview URL などの詳細情報が出力されます。このオプションを設定して、これらの詳細を表示するかどうかを制御できます。

DataFrame 操作のその他の例については、「DataFrame (非推奨)」をご参照ください。

スケジューリングパラメーター

スケジューリングパラメーターの詳細については、「スケジューリングパラメーターの設定と使用」をご参照ください。

実行時パラメーターヒント

o.execute_sql('select * from pyodps_iris', hints={'odps.sql.mapper.split.size': 16})

from odps import options
options.sql.settings = {'odps.sql.mapper.split.size': 16}
o.execute_sql('select * from pyodps_iris')  # グローバル構成に基づいてヒントが追加されます。

サードパーティ製パッケージ

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

他のアカウント

別のアカウントで MaxCompute にアクセスするには、MaxCompute エントリポイントオブジェクトで as_account メソッドを使用します。これにより、指定されたアカウント用の新しいエントリポイントオブジェクトが作成されます。これは、デフォルトの o インスタンスとは独立しています。

トラブルシューティング

実行中にコードが応答しなくなり、出力が生成されない場合は、コードの先頭に次のコメントを追加できます。これにより、DataWorks は 30 秒ごとにすべてのスレッドのスタックトレースを出力します。

# -*- dump_traceback: true -*-

PyODPS のバージョン

現在の PyODPS のバージョンは、Python コードを実行するか、任意の PyODPS タスクのランタイムログを表示することで確認できます。

• PyODPS ノードで次のコードを実行します。

  # 次のコードを入力します：
  import odps; print(odps.__version__)
  # サンプル結果
  0.11.2.3

• PyODPS タスクのランタイムログを確認します。バージョンは、この例のようにログ出力に表示されます。

  2023-07-26 15:39:53  INFO SKYNET_DAGTYPE=100:
  2023-07-26 15:39:53  INFO SKYNET_SYSTEM_ENV=dev:
  2023-07-26 15:39:53  INFO SKYNET_GMTDATE=20230726:
  2023-07-26 15:39:53  INFO SKYNET_BIZDATE=20230726:
  2023-07-26 15:39:53  INFO SKYNET_ENVTYPE=1:
  2023-07-26 15:39:53  INFO SKYNET_PROJECTID=342277:
  2023-07-26 15:39:53  INFO SKYNET_BUSINESS_ID=10756717:
  2023-07-26 15:39:53  INFO SKYNET_ARGS_ENABLE=true:
  2023-07-26 15:39:53  INFO SKYNET_ACCOUNT_ID=xxx
  2023-07-26 15:39:53  INFO TASK_SOURCE=dataworks_datastudio:
  2023-07-26 15:39:53  INFO SKYNET_REGION=xxx:
  2023-07-26 15:39:53  INFO LINK_FILE_ID=503440413:
  2023-07-26 15:39:53  INFO TASK_PLUGIN_NAME=pyodps:
  2023-07-26 15:39:53  INFO ALISA_UNIQUEKEY=xxx
  2023-07-26 15:39:53  INFO ALISA_TASK_ID=T3_xxx
  2023-07-26 15:39:53  INFO ALISA_TASK_EXEC_TARGET=group_599314507759873_dev:
  2023-07-26 15:39:53  INFO ALISA_TASK_PRIORITY=0:
  2023-07-26 15:39:53  INFO --- Invoking Shell command line now ---
  2023-07-26 15:39:53  INFO ============================================================
  Executing user script with PyODPS 0.11.4.post0
  "sepallength","sepalwidth","petallength","petalwidth","name"
  5.4,3.9,1.7,0.4,"Iris-setosa"
  5.4,3.7,1.5,0.2,"Iris-setosa"
  5.8,4.0,1.2,0.2,"Iris-setosa"

付録：ユーザーへの権限付与

別の Alibaba Cloud ユーザー (RAM ユーザーではない) が DataWorks ワークスペース内のプロジェクトやテーブルにアクセスできるようにするには、ODPS SQL ノードを作成し、次の権限付与コマンドを実行する必要があります。ODPS SQL ノードの作成方法の詳細については、「ODPS SQL ノードの作成」をご参照ください。権限の詳細については、「ユーザーと権限」をご参照ください。

-- 別の Alibaba Cloud アカウントユーザーをプロジェクトレベルで追加します。
add user ALIYUN$<account_name>;
-- プロジェクトに対する CreateInstance 権限を付与します。
grant CreateInstance on project <project_name> to USER ALIYUN$<account_name>;
-- テーブルに対する Describe および Select 権限を付与します。
grant Describe, Select on table <table_name> to USER ALIYUN$<account_name>;
-- 権限付与の結果を表示します。
show grants for ALIYUN$<account_name>;

以下は、show grants for ALIYUN$<account_name>; を実行した際の権限付与結果のサンプルです。

Authorization Type: ACL
[user/ALIYUN$:xxx.com]
A    projects/mc_xxx : CreateInstance
A    projects/mc_xxx/tables/pyodps_xxx : Describe | Select

付録：サンプルデータ

この トピック の例に従うには、まずステップ 1: テーブルを作成してデータをインポートするの手順に従って pyodps_iris テーブルを作成し、データを入力してください。このテーブルは基本操作のデモに使用します。

関連トピック

Operating history 機能を使用して、過去 3 日間に DataStudio で実行されたすべてのタスクの履歴を表示します。このビューから、実行中のタスクを停止したり、タスクレコードから SQL ステートメントを一時ファイルとして保存したりできます。詳細については、「操作履歴」をご参照ください。