最適化とクエリプラン

Polars によるデータパイプラインのスケーリングと最適化

Liam Brannigan

Data Scientist & Polars Contributor

クエリ最適化の概要

行と列を持つテーブルの画像

Polars によるデータパイプラインのスケーリングと最適化

クエリ最適化の概要

行と列を持つテーブルの画像

Polars によるデータパイプラインのスケーリングと最適化

クエリ最適化の概要

直列で操作を実行している画像

Polars によるデータパイプラインのスケーリングと最適化

クエリ最適化の概要

並列で操作を実行している画像

Polars によるデータパイプラインのスケーリングと最適化

クエリ最適化の概要

重複する操作が2回実行されるパイプライン

Polars によるデータパイプラインのスケーリングと最適化

クエリ最適化の概要

重複する操作が2回実行されるパイプライン

Polars によるデータパイプラインのスケーリングと最適化

部門別の最多リクエストタイプ

department_request_types = (
    requests





)
Polars によるデータパイプラインのスケーリングと最適化

部門別の最多リクエストタイプ

department_request_types = (
    requests
    .filter(pl.col("STATUS") == "Completed")




)
Polars によるデータパイプラインのスケーリングと最適化

部門別の最多リクエストタイプ

department_request_types = (
    requests
    .filter(pl.col("STATUS") == "Completed")
    .group_by("DEPARTMENT")



)
Polars によるデータパイプラインのスケーリングと最適化

部門別の最多リクエストタイプ

department_request_types = (
    requests    
    .filter(pl.col("STATUS") == "Completed")
    .group_by("DEPARTMENT")
    .agg(pl.col("TYPE").n_unique().alias("n_request_types"))


)
Polars によるデータパイプラインのスケーリングと最適化

部門別の最多リクエストタイプ

department_request_types = (
    requests
    .filter(pl.col("STATUS") == "Completed")
    .group_by("DEPARTMENT")
    .agg(pl.col("TYPE").n_unique().alias("n_request_types"))
    .sort("n_request_types", descending=True)
    .head(5)
)
  • ナイーブプラン
Polars によるデータパイプラインのスケーリングと最適化

最適化前のプラン

print(department_request_types)
Polars によるデータパイプラインのスケーリングと最適化

最適化前のプラン

print(department_request_types)







        Csv SCAN [311_Service_Requests.csv]
        PROJECT */39 COLUMNS
Polars によるデータパイプラインのスケーリングと最適化

最適化前のプラン

print(department_request_types)





      FILTER [(col("STATUS")) == ("Completed")]
      FROM
        Csv SCAN [311_Service_Requests.csv]
        PROJECT */39 COLUMNS
Polars によるデータパイプラインのスケーリングと最適化

最適化前のプラン

print(department_request_types)


    AGGREGATE[maintain_order: false]
      [col("TYPE").n_unique().alias("n_request_types")] BY [col("DEPARTMENT")]
      FROM
      FILTER [(col("STATUS")) == ("Completed")]
      FROM
        Csv SCAN [311_Service_Requests.csv]
        PROJECT */39 COLUMNS
Polars によるデータパイプラインのスケーリングと最適化

最適化前のプラン

print(department_request_types)
SLICE[offset: 0, len: 5]
  SORT BY [descending: [true]] [col("n_request_types")]
    AGGREGATE[maintain_order: false]
      [col("TYPE").n_unique().alias("n_request_types")] BY [col("DEPARTMENT")]
      FROM
      FILTER [(col("STATUS")) == ("Completed")]
      FROM
        Csv SCAN [311_Service_Requests.csv]
        PROJECT */39 COLUMNS
Polars によるデータパイプラインのスケーリングと最適化

最適化後のプラン

print(department_request_types.explain())
Polars によるデータパイプラインのスケーリングと最適化

最適化後のプラン







        Csv SCAN [311_Service_Requests.csv]
        PROJECT 3/39 COLUMNS
        SELECTION: [(col("STATUS")) == ("Completed")]
Polars によるデータパイプラインのスケーリングと最適化

最適化後のプラン






      simple pi 2/2 ["TYPE", "DEPARTMENT"]
        Csv SCAN [311_Service_Requests.csv]
        PROJECT 3/39 COLUMNS
        SELECTION: [(col("STATUS")) == ("Completed")]
Polars によるデータパイプラインのスケーリングと最適化

最適化後のプラン



    AGGREGATE[maintain_order: false]
      [col("TYPE").n_unique().alias("n_request_types")] BY [col("DEPARTMENT")]
      FROM
      simple pi 2/2 ["TYPE", "DEPARTMENT"]
        Csv SCAN [311_Service_Requests.csv]
        PROJECT 3/39 COLUMNS
        SELECTION: [(col("STATUS")) == ("Completed")]
Polars によるデータパイプラインのスケーリングと最適化

最適化後のプラン

SORT BY [slice: (0, 10, ...), descending: [true]] [col("n_request_types")]
  FILTER col("n_request_types").dynamic_predicate() FROM
    AGGREGATE[maintain_order: false]
      [col("TYPE").n_unique().alias("n_request_types")] BY [col("DEPARTMENT")]
      FROM
      simple pi 2/2 ["TYPE", "DEPARTMENT"]
        Csv SCAN [311_Service_Requests.csv]
        PROJECT 3/39 COLUMNS
        SELECTION: [(col("STATUS")) == ("Completed")]
  • n_request_types の上位10行を取得
  • 残りをフィルタリング
  • 10行を並べ替え
Polars によるデータパイプラインのスケーリングと最適化

グラフとして表示する最適化プラン

print(department_request_types.show_graph())

最適化されたプランのグラフビュー(CSVスキャンと最適化を表示)。

Polars によるデータパイプラインのスケーリングと最適化

さらなる最適化

(
    requests








)
Polars によるデータパイプラインのスケーリングと最適化

さらなる最適化

(
    requests
    .filter(pl.col("STATUS") == "Completed")
    .filter(pl.col("DEPARTMENT") == "Sanitation")






)
Polars によるデータパイプラインのスケーリングと最適化

さらなる最適化

(
    requests
    .filter(pl.col("STATUS") == "Completed")
    .filter(pl.col("DEPARTMENT") == "Sanitation")
    .with_columns(
        pl.col("TYPE").str.to_lowercase().alias("type_lower")
    )
    .with_columns(
        pl.col("STATUS").str.to_lowercase().alias("status_lower")
    )
)
Polars によるデータパイプラインのスケーリングと最適化

さらなる最適化



  Csv SCAN [311_Service_Requests.csv]
  PROJECT */39 COLUMNS
  SELECTION: [([(col("DEPARTMENT")) == ("Sanitation")]) & ([(col("STATUS")) == ("Completed")])]
  • AND 述語の結合
Polars によるデータパイプラインのスケーリングと最適化

さらなる最適化

 WITH_COLUMNS:
 [col("TYPE").str.to_lowercase().alias("type_lower"), col("STATUS").str.to_lowercase().alias("status_lower")]
  Csv SCAN [311_Service_Requests.csv]
  PROJECT */39 COLUMNS
  SELECTION: [([(col("DEPARTMENT")) == ("Sanitation")]) & ([(col("STATUS")) == ("Completed")])]
  • AND 述語の結合
  • WITH_COLUMNS 式のクラスタリング
Polars によるデータパイプラインのスケーリングと最適化

練習しましょう!

Polars によるデータパイプラインのスケーリングと最適化

Preparing Video For Download...