DataWorks:PyODPS 3 ノード

はじめに

PyODPS は MaxCompute 向けの Python SDK です。使いやすいプログラミングインターフェイスを提供し、Python を使用してジョブの作成、テーブルやビューのクエリ、MaxCompute リソースの管理を行うことができます。詳細については、「PyODPS」をご参照ください。DataWorks では、PyODPS ノードを使用して Python タスクをスケジュールおよび実行し、他のジョブと統合できます。

制限事項

• 専用スケジューリングリソースグループで PyODPS ノードを実行する場合、ノード内でローカルに処理されるデータ量は 50 MB を超えないようにすることを推奨します。この操作は、専用スケジューリングリソースグループの仕様によって制限されます。過剰な量のローカルデータが処理され、オペレーティングシステムのしきい値を超えると、OOM (Got Killed) エラーが発生する可能性があります。PyODPS ノードで過剰なデータ処理コードを記述することは避けてください。詳細については、「PyODPS を効率的に使用するためのベストプラクティス」をご参照ください。

• サーバーレスリソースグループを使用して PyODPS ノードを実行する場合、処理する必要があるデータ量に基づいて、ノードに適切な数の CU を構成できます。

  説明

  サーバーレスリソースグループでジョブを実行する場合、1 つのジョブに最大 64 CU を構成できます。ただし、ジョブの起動に影響を与える可能性のあるリソース不足を防ぐため、16 CU を超えないようにすることを推奨します。

• Got killed エラーは、プロセスがメモリ制限を超えたために終了したことを示します。したがって、ローカルでのデータ操作は避けてください。PyODPS を介して開始される SQL および DataFrame ジョブ (to_pandas 操作を除く) は、この制限の対象外です。

• プリインストール済みの Numpy および Pandas ライブラリは、ユーザー定義関数を含まないコードで使用できます。バイナリコードを含む他のサードパーティパッケージはサポートされていません。

• 互換性の理由から、DataWorks ではデフォルトで options.tunnel.use_instance_tunnel が False に設定されています。instance tunnel をグローバルに有効にする必要がある場合は、この値を手動で True に設定する必要があります。

• Python 3 のマイナーバージョン (Python 3.8 と Python 3.7 など) の間では、バイトコード定義が異なります。

  MaxCompute は現在 Python 3.7 を使用しています。Python 3.8 の finally block のような他の Python 3 バージョンの構文を使用すると、実行中にエラーが発生します。Python 3.7 の使用を推奨します。

• PyODPS 3 は、サーバーレスリソースグループでの実行をサポートしています。購入して使用するには、「サーバーレスリソースグループの使用」をご参照ください。

• 1 つの PyODPS ノード内で複数の Python ジョブを同時に実行するように構成することはできません。

• PyODPS ノードでログを出力するには、print を使用します。logger.info の使用はサポートされていません。

前提条件

MaxCompute コンピュートエンジンを DataWorks ワークスペースに関連付ける。

操作手順

1. PyODPS 3 ノードのエディターページでコードを開発します。

   PyODPS 3 コード例

   PyODPS ノードを作成した後、コードを編集して実行できます。PyODPS の構文の詳細については、「基本操作」をご参照ください。このトピックでは、5 種類のコード例を紹介します。ビジネスニーズに合ったものを選択できます。

   ODPS エントリポイント

   DataWorks の各 PyODPS ノードには、グローバル ODPS エントリポイント変数 odps または o が含まれています。手動で定義する必要はありません。

   print(odps.exist_table('PyODPS_iris'))

   SQL の実行

   PyODPS ノードで SQL ステートメントを実行できます。詳細については、「SQL」をご参照ください。

   • デフォルトでは、DataWorks で instance tunnel は無効になっています。つまり、instance.open_reader は Result インターフェイスを使用し、最大 10,000 レコードを返します。reader.count を使用してレコード数を取得できます。すべてのデータを反復処理する必要がある場合は、limit を無効にする必要があります。次のステートメントを使用して、instance tunnel をグローバルに有効にし、limit を無効にすることができます。

     options.tunnel.use_instance_tunnel = True
     options.tunnel.limit_instance_tunnel = False  # すべてのデータを読み取るために制限を無効にします。
     
     with instance.open_reader() as reader:
       # すべてのデータは Instance Tunnel を通じて読み取ることができます。

   • また、open_reader 呼び出しに tunnel=True を追加することで、単一の open_reader 呼び出しに対して instance tunnel を有効にすることもできます。limit=False を追加して、その呼び出しの limit 制限を無効にすることもできます。

     # この open_reader 呼び出しに Instance Tunnel を使用し、すべてのデータを読み取ります。
     with instance.open_reader(tunnel=True, limit=False) as reader:

   実行時パラメーターの設定

   • hints パラメーターを使用して実行時パラメーターを設定できます。これは dict 型です。ヒントの詳細については、「SET 操作」をご参照ください。

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

   • sql.settings を使用してグローバル構成を設定すると、関連する実行時パラメーターがすべての実行に追加されます。

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

   実行結果の読み取り

   SQL 実行インスタンスで open_reader 操作を直接呼び出すことができます。これは 2 つのシナリオをサポートします。

   • SQL は構造化データを返します。

     with o.execute_sql('select * from dual').open_reader() as reader:
     	for record in reader:  # 各レコードを処理します。

   • desc などの SQL ステートメントを実行する場合、reader.raw プロパティを使用して SQL 実行の生の結果を取得できます。

     with o.execute_sql('desc dual').open_reader() as reader:
     	print(reader.raw)

     説明

     カスタムスケジューリングパラメーターを使用し、UI から直接 PyODPS 3 ノードを実行する場合、ノードは実行時に変数を置換できないため、時間値をハードコーディングする必要があります。

   DataFrame

   DataFrame (非推奨) を使用してデータを処理することもできます。

   • 実行

     DataWorks 環境では、DataFrame 操作は即時実行メソッドへの明示的な呼び出しが必要です。

     from odps.df import DataFrame
     iris = DataFrame(o.get_table('pyodps_iris'))
     for record in iris[iris.sepal_width < 3].execute():  # 即時実行メソッドを呼び出して各レコードを処理します。

     印刷時に即時実行をトリガーする必要がある場合は、options.interactive を有効にする必要があります。

     from odps import options
     from odps.df import DataFrame
     options.interactive = True  # 最初にオプションを有効にします。
     iris = DataFrame(o.get_table('pyodps_iris'))
     print(iris.sepal_width.sum())  # これにより即時実行がトリガーされます。

   • 詳細情報の出力

     DataWorks では、デフォルトで options.verbose オプションが有効になっており、実行中に Logview URL などの詳細情報が出力されます。

   PyODPS 3 コード開発

   次の簡単な例は、PyODPS ノードの使用方法を示しています。

   1. pyodps_iris サンプルテーブルを作成してデータセットを準備します。詳細については、「DataFrame データの処理」をご参照ください。

   2. DataFrame を作成します。詳細については、「MaxCompute テーブルから DataFrame を作成する」をご参照ください。

   3. PyODPS ノードに次のコードを入力して実行します。

      from odps.df import DataFrame
      
      # ODPS テーブルから DataFrame を作成します。
      iris = DataFrame(o.get_table('pyodps_iris'))
      print(iris.sepallength.head(5))

   PyODPS ジョブの実行

   1. Run Configuration ペインの Compute Resource セクションで、Compute Resource、[コンピューティングクォータ]、および DataWorks Resource Group を構成します。

      説明

      • パブリックネットワークまたは VPC 経由でデータソースにアクセスするには、データソースとの接続性テストに合格したスケジューリングリソースグループを使用する必要があります。詳細については、「ネットワーク接続ソリューション」をご参照ください。

      • ジョブの要件に応じて ミラー 情報を構成できます。

   2. ツールバーのパラメーターダイアログボックスで、作成した MaxCompute データソースを選択し、Run をクリックして PyODPS ジョブを実行します。

2. ノードを定期的に実行するには、ビジネス要件に基づいてスケジューリングプロパティを構成します。詳細については、「ノードスケジューリングの構成」をご参照ください。

   DataWorks の SQL ノードとは異なり、PyODPS ノードはコード内の `${param_name}` のような文字列を置き換えません。代わりに、コードが実行される前に、args という名前の辞書がグローバル変数に追加され、そこからスケジューリングパラメーターを取得できます。たとえば、Parameter セクションで ds=${yyyymmdd} を設定した場合、コード内で次のようにパラメーター情報を取得できます。

   print('ds=' + args['ds'])
   ds=20240930

   説明

   ds という名前のパーティションを取得する必要がある場合は、次のメソッドを使用できます。

   o.get_table('table_name').get_partition('ds=' + args['ds'])

3. ノードが構成された後、デプロイする必要があります。詳細については、「ノードとワークフローのデプロイメント」をご参照ください。

4. ジョブがデプロイされた後、オペレーションセンターでそのステータスを表示できます。詳細については、「オペレーションセンター入門」をご参照ください。

関連付けられたロールでのノードの実行

ノードのリンクされたロールの構成をサポートしており、これにより、特定の RAM ロールでノードタスクを実行して、詳細な権限コントロールとセキュリティ管理を行うことができます。

次のステップ

PyODPS に関するよくある質問：PyODPS の実行中に発生する一般的な問題について学び、例外を迅速に特定して解決します。