MaxCompute:PyODPS

一般的な PyODPS のシナリオ：

• DataFrame クイックスタート。

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

• PyODPS でのパラメーターの受け渡し。

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

• PyODPS でのハッシュパーティションの表示。

• PyODPS での条件付きクエリの実行。

• PyODPS でのシーケンスと実行操作。

PyODPS は、ローカル環境、DataWorks、および PAI ノートブックで実行できます。

• ローカル環境：ローカルに PyODPS をインストールして使用します。ローカル環境での PyODPS の使用。

• DataWorks：DataWorks の PyODPS ノードには PyODPS がプリインストールされており、タスクの開発とスケジューリングに使用できます。DataWorks での PyODPS の使用。

• PAI ノートブック：PAI Python 環境で PyODPS をインストールして実行できます。PyODPS は、PAI-Designer のカスタム Python コンポーネントを含む、PAI の組み込みイメージにもプリインストールされています。使い方は標準的な方法と同じです。基本的な操作およびDataFrame (非推奨)。

PyODPS は、PC、DataWorks の PyODPS ノード、PAI ノートブック環境など、さまざまなクライアントで動作します。 PyODPS は、トンネルダウンロード、execute、to_pandas などの操作を提供し、データをローカルマシンにプルします。新規ユーザーは、データをローカルにプルし、処理してから再度アップロードすることがよくありますが、このアプローチは 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)

  この単一マシンのアプローチでは、データを 1 行ずつ読み取り、処理し、書き込みます。ダウンロードとアップロードのサイクルは遅く、メモリを大量に消費します。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 として実行されます。コアとなる行ごとのロジックは同じままですが、MaxCompute はそれを複数のマシンで並列に実行します。

  • persist インターフェイスは、出力を MaxCompute テーブルに直接書き込むことで、すべてのデータをクラスター内に保持し、ローカルの帯域幅とメモリを節約します。

  • MaxCompute は UDF で jieba のようなサードパーティパッケージをサポートしているため、最小限のコード変更で分散コンピューティングを活用できます。