DataWorks:PyODPS 2 ノード

概要

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

制限事項

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

• サーバーレスリソースグループで PyODPS ノードを実行する場合、処理するデータの量に基づいて PyODPS ノードの CU 割り当てを構成できます。

  説明

  このタスクをサーバーレスリソースグループで実行する場合、単一タスクの最大構成は 64CU ですが、過剰な CU 割り当てによるリソース不足を避け、タスクの起動に影響を与えないように、16CU を超えないようにすることを推奨します。

• Got Killed エラーが発生した場合、メモリ使用量が制限を超え、プロセスが終了したことを示します。したがって、ローカルでのデータ操作は可能な限り避けてください。PyODPS を介して開始された SQL および DataFrame タスク (to_pandas を除く) は、この制限の対象外です。

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

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

• PyODPS 2 ノードの基盤となる Python のバージョンは 2.7 です。

• PyODPS ノード内で複数の Python タスクを同時に実行することはサポートされていません。

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

前提条件

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

操作手順

1. PyODPS 2 ノードの編集ページで、次の開発操作を実行します。

   PyODPS 2 コードの例

   PyODPS ノードを作成した後、コードを編集して実行できます。PyODPS の構文に関する詳細については、「PyODPS ドキュメント」をご参照ください。このトピックでは、次の 5 つのコード例について説明します。ビジネスニーズに基づいて例を選択できます。

   ODPS エントリ

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

   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  # limit 制限を無効にし、すべてのデータを読み取ります。
     
     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:

   Instance Tunnel を有効にしない場合、取得したデータ形式が正しくない可能性があります。解決策については、「データ形式エラーのトラブルシューティング」をご参照ください。

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

   • hints パラメーターを構成することで、実行時パラメーターを設定できます。このパラメーターは dict 型です。hints パラメーターの詳細については、「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:  # 各レコードを処理します。

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

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

     説明

     カスタムスケジューリングパラメーターを使用する場合、PyODPS ノードは直接パラメーターを置き換えることができないため、ページ上で PyODPS 2 ノードを直接トリガーする際には時間値をハードコーディングする必要があります。

   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():  # 即時実行メソッドを呼び出し、各 Record を処理します。

     Print を使用する際に即時実行をトリガーしたい場合は、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())  # Print 時に即時実行されます。

   • 詳細情報の出力

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

   PyODPS 2 コード開発

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

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

   2. DataFrame を作成します。詳細については、「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、コンピュートリソースとクォータ、およびDataWorks Resource Groupを構成します。

      説明

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

      • タスクの要件に基づいてミラー情報を構成できます。

   2. ツールバーで 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 ロールを関連付けて、特定の RAM ロールでノードタスクを実行し、詳細な権限管理とセキュリティ管理を有効にすることができます。

次のステップ

PyODPS のよくある質問：PyODPS の実行中によくある問題について学び、迅速なトラブルシューティングに役立てることができます。