MaxCompute:PyODPS

PyODPS の使用シナリオの詳細については、以下のトピックをご参照ください。

• はじめに

• PyODPS ノードを使用してパーティションテーブルからデータを読み取る

• PyODPS ノードを使用してパラメーターを渡す

• PyODPS ノードでサードパーティパッケージを参照する

• PyODPS ノードを使用して、指定されたテーブルのレベル 1 パーティションからデータを読み取る

• PyODPS ノードを使用して特定の条件に基づいてデータをクエリする

• PyODPS ノードを使用してシーケンス操作を実行する

PyODPS は、オンプレミス環境、DataWorks、および Machine Learning Platform for AI (PAI) Notebooks で実行できます。

• オンプレミス環境: オンプレミス環境に PyODPS をインストールして使用できます。詳細については、オンプレミス環境で PyODPS を使用するをご参照ください。

• DataWorks: PyODPS は、DataWorks の PyODPS ノードにインストールされています。DataWorks [コンソール] で PyODPS ノードを開発し、ノードを定期的に実行できます。詳細については、DataWorks で PyODPS を使用するをご参照ください。

• PAI Notebooks: PAI の Python 環境にも PyODPS をインストールして実行できます。PyODPS は、Machine Learning Designer のカスタム Python コンポーネントなど、PAI の組み込みイメージにインストールされており、直接使用できます。PAI Notebooks では、PyODPS に関連する基本操作を実行できます。詳細については、基本操作の概要とDataFrame の概要をご参照ください。

すべてのデータをオンプレミスマシンにダウンロードして PyODPS を実行しないでください。PyODPS は SDK であり、PC、DataWorks (DataStudio の PyODPS ノード)、PAI Notebooks など、さまざまなクライアントで実行できます。 PyODPS を使用すると、さまざまな方法でデータをオンプレミスマシンにダウンロードできます。たとえば、Tunnel コマンド、execute コマンド、または to_pandas メソッドを使用してデータをダウンロードできます。そのため、最初に PyODPS を使用する多くのユーザーは、データをオンプレミスマシンにダウンロードし、データが処理された後に MaxCompute にアップロードしようとします。ほとんどの場合、このデータ処理方法は非効率的であり、MaxCompute の大規模並列計算機能を活用していません。

次の例では、トークン化のための 2 つのデータ処理方法のコードを比較します。

• シナリオ

  毎日生成されるログ文字列を分析して特定の情報を抽出します。1 つの列のみを含むテーブルが存在し、その列は STRING データ型です。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 の大規模計算機能を使用できます。