データ ファイルをパーティション分割する

  • 5 分

パーティション分割は、Spark がワーカー ノード全体のパフォーマンスを最大化できるようにする最適化の手法です。 不要なディスク IO を排除して、クエリ内のデータをフィルター処理すると、パフォーマンスがより一層向上します。

出力ファイルをパーティション分割する

データフレームをパーティション分割されたファイルのセットとして保存するには、データの書き込み時に partitionBy メソッドを使用します。

次の例では、派生した Year フィールドを作成します。 次に、それを使用してデータをパーティション分割します。

from pyspark.sql.functions import year, col

# Load source data
df = spark.read.csv('/orders/*.csv', header=True, inferSchema=True)

# Add Year column
dated_df = df.withColumn("Year", year(col("OrderDate")))

# Partition by year
dated_df.write.partitionBy("Year").mode("overwrite").parquet("/data")

データフレームのパーティション分割時に生成されるフォルダー名には、ここに示すように、パーティション分割列の名前と値が column=value 形式で含まれます。

Diagram representing a partitioned file folder structure.

Note

データは複数の列でパーティション分割できます。これにより、パーティション キーごとにフォルダーの階層が作成されます。 たとえば、この例の順序を年と月でパーティション分割して、フォルダー階層に各年の値のフォルダーが含まれ、さらに各月の値のサブフォルダーが含まれるようにすることができます。

クエリで Parquet ファイルをフィルター処理する

Parquet ファイルからデータフレームにデータを読み取る場合は、階層フォルダー内の任意のフォルダーからデータをプルすることができます。 このフィルター処理のプロセスは、パーティション分割されたフィールドに対して明示的な値とワイルドカードを使用して行われます。

次に示す例では、次のコードは、2020 年に発注された販売注文をプルします。

orders_2020 = spark.read.parquet('/partitioned_data/Year=2020')
display(orders_2020.limit(5))

Note

ファイル パスで指定されたパーティション分割列は、結果のデータフレームでは省略されます。 この例のクエリによって生成される結果には Year 列は含まれません。すべての行は 2020 からのものになります。