MaxCompute:PyODPS の概要

PyODPS は以下のシナリオで使用されます。

• クイックスタート

• PyODPS を使用したパーティションテーブルからのデータ読み取り

• PyODPS でのパラメータ渡し

• PyODPS でのサードパーティパッケージの使用

• PyODPS を使用したハッシュパーティションの表示

• PyODPS を使用した条件クエリの実行

• PyODPS での Sequence と実行操作の使用

PyODPS は、ローカル環境、DataWorks、および PAI Notebook で実行できます。

• ローカル環境: ローカル環境に PyODPS をインストールして使用できます。詳細については、「ローカル環境での PyODPS の使用」をご参照ください。

• DataWorks: DataWorks の PyODPS ノードには PyODPS がプリインストールされています。これらのノードで PyODPS ジョブを直接開発し、定期的に実行できます。詳細については、「DataWorks での PyODPS の使用」をご参照ください。

• PAI Notebook: PAI Python 環境に PyODPS をインストールして実行できます。PyODPS は、PAI-Designer のカスタム Python コンポーネントなど、組み込みの PAI イメージにプリインストールされており、すぐに使用できます。PAI Notebook での PyODPS の使用方法は、標準的な使用方法と同様です。詳細については、「基本操作の概要」および「データフレーム (非推奨)」をご参照ください。

PyODPS は、PC、DataWorks の PyODPS ノード、PAI Notebook 環境など、さまざまなクライアント上で実行される SDK です。 PyODPS は、トンネルダウンロード、実行、to_pandas といった、データをローカルマシンに取得するための便利な操作をいくつか提供しています。その結果、多くの新規ユーザーは、データをローカルに取得して処理してから MaxCompute にアップロードし直そうと試みます。しかし、この方法は多くの場合、非常に非効率的です。データをローカルに取得すると、MaxCompute の大規模並列計算機能を活用できなくなります。

以下の形態素解析の例では、両方の方法のコードを比較します。

• シナリオ例：

  日次ログ文字列を分析して情報を抽出する必要があります。文字列型の単一列を含むテーブルがあります。jieba ライブラリを使用して中国語テキストを形態素解析し、必要なキーワードを見つけて、情報テーブルに保存する必要があります。

• 非効率的な処理コードのデモ：

  import jieba
  t = o.get_table('word_split')
  out = []
  with t.open_reader() as reader:
      for r in reader:
          words = list(jieba.cut(r[0]))
          #
          # processed_data を生成する処理ロジック
          #
          out.append(processed_data)
  out_t = o.get_table('words')
  with out_t.open_writer() as writer:
      writer.write(out)

  このアプローチは、単一マシン処理という考え方に基づいています。つまり、データを行ごとに読み取り、処理し、宛先テーブルに書き込みます。プロセス全体で、データのダウンロードとアップロードに長時間を要します。スクリプトを実行するマシンも、すべてのデータを処理するために大量のメモリを必要とします。DataWorks ノードのユーザーにとって、このアプローチは、デフォルトで割り当てられたメモリを超え、OOM エラーを引き起こしやすくなります。

• 効率的な処理コードのデモ：

  from odps.df import output
  out_table = o.get_table('words')
  df = o.get_table('word_split').to_df()
  
  # 返す必要のあるフィールドと型を想定
  out_names = ["word", "count"]
  out_types = ["string", "int"]
  
  @output(out_names, out_types)
  def handle(row):
      import jieba
      words = list(jieba.cut(row[0]))
      #
      # processed_data を生成する処理ロジック
      #
      yield processed_data
  df.apply(handle, axis=1).persist(out_table.name)

  apply メソッドを使用した分散実行：

  • 複雑なロジックは handle 関数に記述します。この関数は自動的にシリアル化されてサーバー側に送られ、UDF として呼び出され、実行されます。handle 関数もサーバー側の実行中にデータを行ごとに処理するため、ロジックは同一です。違いは、このプログラムが MaxCompute にサブミットされて実行されると、複数のマシンが同時にデータを処理する点です。これにより、処理時間を大幅に短縮できます。

  • persist インターフェイスを呼び出すと、生成されたデータが別の MaxCompute テーブルに直接書き込まれます。すべてのデータの生成と消費は MaxCompute クラスター内で完結します。これにより、ローカルのネットワークリソースとメモリリソースが節約されます。

  • この例では、サードパーティパッケージも使用しています。MaxCompute では、UDF でこの例の jieba のようなサードパーティパッケージを使用できます。そのため、コード変更のコストを心配する必要はありません。主要なロジックをほとんど変更せずに、MaxCompute の大規模コンピューティング機能を活用できます。