Optimering och frågeplaner

Skalning och optimering av datapipelines med Polars

Liam Brannigan

Data Scientist & Polars Contributor

Introduktion till frågeoptimering

Bild av tabell med rader och kolumner

Introduktion till frågeoptimering

Bild av tabell med rader och kolumner

Introduktion till frågeoptimering

Bild av operationer som körs seriellt

Introduktion till frågeoptimering

Bild av operationer som körs parallellt

Introduktion till frågeoptimering

Pipeline med duplicerade operationer som utförs två gånger

Introduktion till frågeoptimering

Pipeline med duplicerade operationer som utförs två gånger

Flest ärendetyper per avdelning

department_request_types = (
    requests





)

Flest ärendetyper per avdelning

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




)

Flest ärendetyper per avdelning

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



)

Flest ärendetyper per avdelning

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


)

Flest ärendetyper per avdelning

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)
)

Naiv plan

Ooptimerad plan

print(department_request_types)

Ooptimerad plan

print(department_request_types)








        Csv SCAN [311_Service_Requests.csv]
        PROJECT */39 COLUMNS

Ooptimerad plan

print(department_request_types)






      FILTER [(col("STATUS")) == ("Completed")]
      FROM
        Csv SCAN [311_Service_Requests.csv]
        PROJECT */39 COLUMNS

Ooptimerad plan

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

Ooptimerad plan

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

Optimerad plan

print(department_request_types.explain())

Optimerad plan







        Csv SCAN [311_Service_Requests.csv]
        PROJECT 3/39 COLUMNS
        SELECTION: [(col("STATUS")) == ("Completed")]

Optimerad plan






      simple pi 2/2 ["TYPE", "DEPARTMENT"]
        Csv SCAN [311_Service_Requests.csv]
        PROJECT 3/39 COLUMNS
        SELECTION: [(col("STATUS")) == ("Completed")]

Optimerad plan



    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")]

Optimerad plan

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")]

Hitta de 10 översta raderna i n_request_types
Filtrera bort resten
Sortera de 10 raderna

Optimerad plan som graf

print(department_request_types.show_graph())

Grafvy över den optimerade planen med CSV-skanning och optimeringar.

Ytterligare optimeringar

(
    requests








)

Ytterligare optimeringar

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






)

Ytterligare optimeringar

(
    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")
    )
)

Ytterligare optimeringar



  Csv SCAN [311_Service_Requests.csv]
  PROJECT */39 COLUMNS
  SELECTION: [([(col("DEPARTMENT")) == ("Sanitation")]) & ([(col("STATUS")) == ("Completed")])]

Kombinerat AND-predikat

Ytterligare optimeringar

 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")])]

Kombinerat AND-predikat
Grupperade WITH_COLUMNS-uttryck

Nu kör vi en övning!

Skalning och optimering av datapipelines med Polars