MaxCompute:PyODPS 3 タスクの開発

前提条件

PyODPS 3 ノードが作成されていること。詳細については、「MaxCompute ノードの作成と管理」をご参照ください。

背景情報

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

制限事項

• 専用スケジューリングリソースグループで PyODPS ノードを実行する場合、50 MB を超えるローカルデータを処理しないでください。これは、専用スケジューリングリソースグループのリソース仕様によるものです。オペレーティングシステムのしきい値を超える大量のローカルデータを処理すると、Got Killed メッセージが表示され、メモリ不足 (OOM) エラーが発生する可能性があります。PyODPS ノードに大量のデータ処理コードを直接記述しないでください。

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

  説明

  サーバーレスリソースグループでタスクを実行する場合、単一タスクは最大 64 CU の設定をサポートしますが、過剰な CU 値によるリソース不足を防ぐため、16 CU を超えないことを推奨します。これはタスクの起動に影響を与える可能性があります。

• [Got killed] エラーは、メモリ使用量が制限を超え、プロセスが終了したことを示します。これを防ぐには、ローカルデータ操作を避けてください。この制限は、PyODPS を通じて開始される SQL またはデータフレーム タスク (to_pandas を除く) には適用されません。

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

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

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

  MaxCompute は現在 Python 3.7 を使用しています。Python 3.8 の finally ブロックなど、他の Python 3 バージョンの構文を使用すると、実行エラーが発生します。Python 3.7 を使用することを推奨します。

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

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

コードの編集：基本的な例

PyODPS ノードを作成した後、コードを編集して実行できます。PyODPS の構文の詳細については、「基本操作の概要」をご参照ください。

• ODPS エントリポイント

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

  print(odps.exist_table('PyODPS_iris'))

• SQL の実行

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

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

    options.tunnel.use_instance_tunnel = True
    options.tunnel.limit_instance_tunnel = False  # すべてのデータを読み取るために limit を無効にします。
    with instance.open_reader() as reader:
      # インスタンストンネルを通じてすべてのデータを読み取ることができます。

  • また、tunnel=True を open_reader に追加して、現在の open_reader 呼び出しに対してインスタンストンネルを有効にすることもできます。同様に、limit=False を追加して、現在の呼び出しの limit 制限を無効にできます。

    # 現在の open_reader 操作に対して Instance Tunnel インターフェイスを使用し、すべてのデータを読み取ります。
    with instance.open_reader(tunnel=True, limit=False) as reader:

• ランタイムパラメータ

  • 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')  # この呼び出しには、グローバル設定の hints が含まれます。

• 実行結果

  SQL 実行インスタンスは、次の 2 つのシナリオで open_reader 操作を直接実行できます。

  • SQL ステートメントが構造化データを返す場合。

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

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

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

    説明

    カスタムスケジューリングパラメータを使用する場合、ページから PyODPS 3 ノードを直接トリガーして実行する際には、時間をハードコードする必要があります。PyODPS ノードはこの値を直接置換できません。

• データフレーム

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

  • 実行

    DataWorks 環境では、データフレーム操作は、即時実行メソッドを呼び出すことによって明示的にトリガーする必要があります。

    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())  # 出力時に即時実行がトリガーされます。

  • 詳細情報の出力

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

例

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

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

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

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

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

   次の結果が返されます。

      sepallength
   0          4.5
   1          5.5
   2          4.9
   3          5.0
   4          6.0

コードの編集：高度な例

ノードを定期的に実行する必要がある場合は、そのスケジューリングプロパティを定義する必要があります。詳細については、「ノードのスケジューリングプロパティの設定」をご参照ください。

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

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

他のユースケースでの PyODPS タスクの開発の詳細については、次のトピックをご参照ください。

• 「PyODPS でサードパーティパッケージを使用する」

• 「PyODPS ノードで Jieba を使用して中国語テキストをセグメント化する」

• 「PyODPS ノードを使用してローカルコンピュータへのデータダウンロードを回避する」

次のステップ

• 「カスタム Shell スクリプトが正常に実行されたかどうかを判断する」：カスタム Python スクリプトが正常に実行されたかどうかを判断するロジックは、Shell スクリプトの場合と同じです。この方法を検証に使用できます。

• 「タスクのデプロイ」：標準モードのワークスペースを使用している場合、定期的に実行する前に、タスクを本番環境にデプロイする必要があります。

• 「定期実行タスクの運用保守」：タスクが本番環境にデプロイされ、スケジュールされた後、オペレーションセンターでタスクの運用保守を実行できます。

• 「PyODPS FAQ」：PyODPS タスクの実行に関する一般的な質問への回答を見つけて、問題を迅速にトラブルシューティングできます。

よくある質問

Q： PyODPS 3 ノードを使用して、Lark などのサードパーティ API からデータを収集し、DataWorks にインポートしています。コードはローカル開発環境では問題なく実行されますが、本番環境に送信してオペレーションセンターで実行すると、応答タイムアウトエラーが発生します。なぜですか？

A： 管理センター > セキュリティ設定 > ワークスペース。サンドボックスホワイトリストで、サードパーティ API のドメイン名を追加して、PyODPS 3 タスクにアクセスを許可してください。例：

Lark API ドメイン名 open.feishu.cn を追加し、ポートを 443 に設定します。