PySpark トランスフォーム

トランスフォームの入力と出力、軽量トランスフォームと PySpark トランスフォームの違いなど、基本的な概念については、トランスフォームを参照してください。

PySpark は、Apache Spark バックエンドと連携してデータを高速に処理できるラッパー言語です。Spark は、サーバーの分散ネットワーク全体で非常に大規模なデータセットを処理でき、正しく使用すれば、パフォーマンスと信頼性の面で大きなメリットがあります。ただし、特に SQL などのリレーショナルデータベースシステムに慣れている場合は、いくつかの制約もあります。たとえば、Spark は、ある行がどのサーバーのどの位置に存在するかを正確に把握できないため、特定の行を直接選択して更新または削除することはできません。このような方法でデータベースを扱うことに慣れている場合は、データセットを全体として捉え、行ではなく列に基づいてデータを処理するように、概念モデルを調整する必要があります。

ループ、リスト、変数など、手続き型の Python の構文に慣れている場合は、Python から移行するを参照して、PySpark の DataFrame オブジェクトの使い方を学んでください。

• DataFrame： 名前付きの列の下に行をまとめたものです。
  • 構造は SQL データベースに似ていますが、リレーショナルではありません。
  • 不変： DataFrame は作成後に変更できませんが、新しい DataFrame に変換できます（その結果、元の DataFrame と変換後の DataFrame の2個が存在します）。データセットは上書きできますが、Foundry はバージョン履歴を保持するため、いつでも過去のビルドを探索し、そのビルドに戻ることができます。
  • 遅延評価： 一連の変換タスクは、1つの（組み合わせた）アクションとして評価され、ビルドがトリガーされたときに実行されます。
• Resilient Distributed Datasets： (RDD) は、DataFrame の基盤となるデータ構造です。DataFrame を互いに重ならない複数の部分集合にパーティション分割することで、コンピューターのクラスター（ネットワーク）内にある複数のコンピューター（ノード）上で変換を並列に評価できます。これらはすべて内部で行われますが、PySpark でコードを書く際には、この仕組みを念頭に置くことが重要です。
• 共有変数： デフォルトでは、Spark は効率を高めるため、変換タスクで使用する各変数の個別に管理されるコピーを、並列処理を行う各コンピューター（ノード）に送信します。タスク間で変数を共有する必要がある場合、Spark は2種類の共有変数をサポートしています。
  1. ブロードキャスト変数： クラスター内のすべてのコンピューター（ノード）にブロードキャストされる値を、メモリー (RAM) にキャッシュ（保存）します。
  2. アキュムレーター： 加算または集計できる変数で、カウンターや合計などが含まれますが、これらに限定されません。この概念は GroupedData に関連し、統計計算に役立ちます。
• DataFrame を使用する理由： Spark の DataFrame は、大量（ペタバイト以上）の構造化データを処理するために設計され、最適化されています。
• PySpark コードを書く理由： PySpark を使用すると、Code Repositories と Code Workbook でデータセットの加工方法をカスタマイズでき、Contour や Blacksmith だけを使用する場合よりも複雑かつ柔軟な加工ができます。
• PySpark が適していない用途： PySpark はデータセットの加工を目的として設計されており、個々の値へのアクセスを目的としていません。合計や平均を計算することはできますが、データを直接参照することはできず、そうするべきでもありません。

クエリの結果が「ビュー」（仮想テーブルの結果セット）になる SQL とは異なり、PySpark でデータセットを処理すると、まったく新しいデータセットが生成されます。これにより、派生データセットを基に新しいデータセットをビルドできるだけでなく、組織内のほかのメンバーも、中間データセットをご自身のデータ処理タスクに再利用できます。データオペレーティングシステムである Palantir Foundry では、データセットは親子（またはソースと結果）の有向ツリー関係によって自動的にリンクされます。これにより、誰でも Spark による変換のデータリネージを簡単にたどれます。つまり、データセットの依存関係がどのようにビルドされ、それらのデータセットがどこから来たのかを探索できます。また、組織内のほかのメンバーがデータセットをどのように使用しているかも見つけられるため、例から学んだり、重複する作業を効果的に減らしたりできます。

PySpark コードを理解する

スターターコードの基本

Code Workbook では、関数は次のようになります。

Copied!
1
2
3
4
def new_frame(old_frame):
  df = old_frame
  # df = transformations on df
  return df

• old_frame：Foundry 内に保存されたデータセットを表す DataFrame を参照します。old_frame は不変であり、この new_frame 関数内では変更できません。ある意味では、変換の中間ステップはすべて、新しい不変のデータフレームを生成します。そのデータフレームをさらに加工したり、そのまま返したりできます。これは完全に正確な説明ではありませんが、考え方のモデルとして、コードをより整理しやすくなります。
• new_frame：この関数内で、old_frame に適用する一連の変換を定義できます。return 文では DataFrame（この例では df と呼んでいます）を返す必要があります。内部では、その DataFrame に適用したすべての変換がまとめられ、最適化されてから、入力データセットに適用されます。コードでビルドをトリガーすると、結果は Foundry 内の新しいデータセットファイルに保存され、ビルドの完了後に探索できます。

DataFrame 内のデータは、Array でも Dictionary でもないため、直接参照できません。実際には、内部でパーティション分割やシャッフルが行われているため、どの時点でもデータの所在を特定することはできません。データセットの絞り込みや集計を行う場合を除き、コードはデータセットの内容にあまり依存しないものにする必要があります。並べ替えは一般にコストが高く、時間がかかるため、基本的には、行がランダムな順序で並んでいると想定し、列、フィルター、集計、そして創意工夫による問題解決に手段を絞ります。

名前付き列

DataFrame の各列には名前が付いており、名前を変更できます。列名は一意で、大文字と小文字が区別されます。Foundry データセットでは、次のガイドラインに従ってください。

• 常に小文字または数字を使用します。
• 単語はスペースではなく _（アンダースコア）で区切ります（スペースは使用できません）。
• 命名規則として、camelCasedColumnNames は避けます。
• (、)、& などの特殊文字は使用しないでください。
• 集計関数は、特殊文字を含む列名を自動的に付けることがあります。トランスフォームでデータフレームを返す前に、エイリアスを指定するか、列名を変更する必要があります。

トランスフォームを連結する

既存のコードを見ると、DataFrame を参照する変数の命名方法に厳密なルールがないことに気づくでしょう。このチートシートでは、DataFrame を df として参照しますが、ほかの例では raw、out、input、table、something_specific などが使われることもあります。目的を果たせるのであれば、どの名前でも構いません。

また、次のようなパターンも見られます。

Copied!
1
2
3
4
5
df = df.select("firstName", "age")
df = df.withColumn("age", df.age.cast("integer"))
df = df.filter(df.age > 21)
df = df.withColumnRenamed("firstName", "first_name")
return df

または、同じ処理を別の書き方で記述できます。

Copied!
1
2
3
4
return df.select("firstName", "age") \
       .withColumn("age", df.age.cast("integer")) \
       .filter(df.age > 21) \
       .withColumnRenamed("firstName", "first_name")

コーディングに慣れていない方のために説明すると、= の左側にある df には、右側の df に適用された変換の結果が格納されてから、コードの次の行に進みます。この例では、結果を同じ名前の変数に格納しており、各ステップの後で df の内容を実質的に上書きしています。DataFrame の変換結果を別の名前の変数に格納することもできますが、ほとんどの場合は、同じ変数を上書きして次に進んでも問題ありません。各トランスフォーム関数の最後では、新しいデータフレームを、変数として（1つ目の例）、または最後の変換の結果として（2つ目の例）返す必要があります。

どちらの例も同じ処理を行います。

1. 加工後のデータセットに含める df の2列のみを選択します。
2. age 列をキャストして、文字列ではなく Integer にします。
3. データセットの行を絞り込み、age > 21 のエントリのみを含めます。
4. 列名を firstName から first_name に変更します。

結果のデータセットには first_name と age の2列のみが含まれ、21歳以下の人は除外されます。これが最終的な df の内容であり、return で返すことも、さらに変換を適用することもできます。以降のセクションでは、これらのトランスフォームを詳しく説明します。

ウィンドウ関数

ウィンドウ関数は、DataFrame オブジェクトの元の行や列を削除せずに、集計値やその他の計算値を追加します。累計、順位付け、移動平均に使用します。詳しくは、ウィンドウ関数を参照してください。

Foundry で PySpark を記述する

Foundry で PySpark を記述するためのツールには、Code Repositories と Code Workbook の2つがあります。

Code Repositories では、ほとんどの関数を使用するために、.py ドキュメントの先頭で次のインポート文を宣言する必要があります。

Copied!
1
from pyspark.sql import functions as F

Code Workbook では、これがグローバルインポートとしてすでに含まれているため、追加の設定なしでほとんどの関数を使用できます。

このリファレンスは網羅的なものではなく、一般的なパターンやベストプラクティスに関するガイダンスに重点を置いています。PySpark SQL 関数の完全な一覧については、公式の Apache Spark ドキュメント ↗を参照できます。

明快で読みやすく、保守しやすい PySpark コードを記述するためのガイダンスについては、PySpark スタイルガイドを参照してください。