Chapter 3. Eager and Lazy APIs

Overview of Polars APIs

• Two types of Polars APIs:
  1. Eager API
  2. Lazy API
• Each API serves different use cases and has distinct performance characteristics.

Eager API

• Immediate execution model:
  • Functions are executed sequentially.
  • Data manipulation happens in real time.
• Use cases:
  • Ideal for data exploration and iterative tasks.
  • Provides immediate feedback after each operation.
  • Similar experience to Pandas, making it easier for users familiar with Pandas to transition.

Lazy API

• Deferred execution model:
  • Execution of data transformations is postponed until required.
• Performance benefits:
  • Allows Polars to optimize queries before execution.
  • Particularly useful in large-scale and performance-sensitive scenarios.

Eager API: DataFrame

• Eager API operates on an immediate execution model, executing each function sequentially.
• Ideal for data exploration and iterative analysis, providing direct interaction with data.
• Similar to Pandas, making it intuitive for users transitioning from Pandas.

Example: Taxi Trip Analysis with Eager API

Goal: Analyze taxi trip data to determine the top three vendors by revenue per distance traveled.

%%time
trips = pl.read_parquet("data/taxi/yellow_tripdata_*.parquet")
sum_per_vendor = trips.group_by("VendorID").sum()

income_per_distance_per_vendor = sum_per_vendor.select(
    "VendorID",
    income_per_distance=pl.col("total_amount") / pl.col("trip_distance")
)

top_three = (
    income_per_distance_per_vendor.sort(
        by="income_per_distance",
        descending=True
    )
    .head(3)
)
top_three

1. Read Data:

   • Loads all taxi trip data from matching Parquet files into a single DataFrame.

2. Group and Sum:

   • Groups data by vendor and calculates sums of key columns like total amount and trip distance.

3. Calculate Metrics:

   • Computes income per distance for each vendor by dividing the total income by the total distance traveled.

4. Sort and Select:

   • Ranks vendors by their income per distance and extracts the top three vendors.

Benefits of Eager API in Analysis

• Tackling problems in smaller steps provides immediate feedback, allowing you to inspect the data after each step.
• This step-by-step approach helps make informed decisions about the next operations, improving analysis accuracy.

LazyFrame

lazy_df = pl.LazyFrame({
    "foo": [1, 2, 3, 4, 5],
    "bar": [6, 7, 8, 9, 10]
})

1. Common Subplan Elimination

• Definition: A subplan (or subtree) is a group of steps in the query plan. If multiple operations share the same subplan, the results are cached and reused, avoiding redundant computation.

• Example:

  common_subplan = lazy_df.with_columns(pl.col("foo") * 2)
  
  expr1 = common_subplan.filter(pl.col("foo") * 2 > 4)
  expr2 = common_subplan.filter(pl.col("foo") * 2 < 8)
  
  result = pl.concat([expr1, expr2])

• Before optimization: The file is read twice for each expression.

• After optimization: The file is read only once, with the shared subplan reused for both expressions.

• Graph visualization:

  • Use result.show_graph(optimized=False) to view the unoptimized plan.
  • Use result.show_graph() to view the optimized plan.

2. Eager API Internally Using Lazy API

• Eager API often calls the lazy API under the hood:
  • Allows the query planner to optimize queries before execution.
  • Simplifies the codebase, as eager methods act as thin wrappers around lazy API methods.

3. Schema Error Detection

• Lazy API detects schema errors before data is processed, allowing fast failure and quick feedback.

• Example: Treating an integer column as a string raises an immediate SchemaError.

  ldf = pl.LazyFrame({
      "name": ["Alice", "Bob", "Charlie"],
      "age": [25, 30, 35]
  })
  
  erroneous_query = ldf.with_columns(
      pl.col("age").str.slice(1, 3).alias("sliced_age")
  )
  
  result_df = erroneous_query.collect()
  # Output: SchemaError: invalid series dtype: expected `String`, got `i64`

  • Benefit: Prevents long-running queries from failing late due to schema mismatches, improving programming efficiency with instant feedback.

Key Takeaways

• Common subplan elimination avoids redundant data reading by reusing shared subplans.
• Eager API often leverages lazy API optimizations, ensuring efficient execution.
• Lazy API helps catch schema errors early, preventing costly errors in large, long-running queries.

Performance Differences: Eager API vs. Lazy API

Compared to previous snippets ... Lazy API Query Example:

%%time
trips = pl.scan_parquet("data/taxi/yellow_tripdata_*.parquet")
sum_per_vendor = trips.group_by("VendorID").sum()
income_per_distance_per_vendor = sum_per_vendor.select(
    "VendorID",
    income_per_distance=pl.col("total_amount") / pl.col("trip_distance")
)
top_three = income_per_distance_per_vendor.sort(
    by="income_per_distance",
    descending=True
).head(3)
top_three.collect()

• The lazy API is about 10 times faster than the eager API in this scenario.
  • Reason: The lazy API defers execution and performs query optimizations, leading to faster results.

LazyFrame and collect() Method

• LazyFrame is only evaluated when you call **collect()**.
  • This triggers the conversion of the LazyFrame into a DataFrame.
• However, repeated calls to collect() will recompute the LazyFrame from scratch, re-running the entire computation.

Example: Repeated Computation

• Creating a LazyFrame and performing computations:

  lf = pl.LazyFrame({"col1": [1, 2, 3], "col2": [4, 5, 6]})
  print(lf.collect())
  print(lf.with_columns(pl.col("col1") + 1).collect())  # Recomputes the LazyFrame

• Note: The second collect() recalculates the entire LazyFrame, leading to repeated computation.

Key Takeaways

• Lazy API offers substantial performance benefits over the eager API, especially for complex queries and large datasets.
• Recomputing a LazyFrame via multiple collect() calls can lead to inefficiencies, as it re-executes the query from scratch.

Functionalities Difference Between LazyFrame and DataFrame

• LazyFrame: Data is not available until it’s collected using the collect() method.
• DataFrame: Data is available immediately, allowing all operations to be performed directly.

Aggregations in LazyFrame vs. DataFrame

• Common Aggregations:

  • Both LazyFrame and DataFrame support column-wise aggregations, such as:
    • .max()
    • .mean()
    • .median()
    • .min()
    • .sum()
    • .std()
    • .var()
    • .null_count()
    • .quantile()
    • .product()
  • In LazyFrame, these operations are deferred until data is collected.

• Horizontal Aggregations (DataFrame-Only):

  • Horizontal aggregations apply operations row-wise across columns (e.g., summing multiple columns for each row).
  • Available only in DataFrame, not in LazyFrame.
  • Examples of horizontal aggregations:
    • .max_horizontal()
    • .mean_horizontal()
    • .min_horizontal()
    • .sum_horizontal()

Attribute Differences Between DataFrame and LazyFrame

• Missing Attributes in LazyFrame:
  • LazyFrame does not have .shape, .height, and .flags attributes because:
    • Shape/Height: Number of rows and columns is only known when the data is collected.
    • Flags: Used internally for optimizations like indicating if a column is sorted.

Computation Methods

• DataFrame:

  • Supports computation methods like:
    • .fold(): Row-wise reduction by applying a function that reduces two Series to one.
    • .hash_rows(): Hashes each row to a UInt64 value.

• LazyFrame:

  • Does not support row-wise computation methods like .fold() and .hash_rows().

Descriptive Methods Differences

• LazyFrame:

  • Has only two descriptive methods:
    • .explain(): Showcases the query plan.
    • .show_graph(): Visualizes the query graph.

• DataFrame:

  • Provides a range of descriptive methods to analyze the data:
    • .describe(): Summarizes statistics about each column.
    • .estimated_size(): Estimates the memory size of the DataFrame.
    • .is_duplicated(), .is_empty(), .is_unique(): Check for duplicates, emptiness, or uniqueness.
    • .n_unique(), .approx_n_unique(): Get the number of unique values.
    • .glimpse(): Offers a quick view of the data.

Computation Methods

• DataFrame:

  • Supports computation methods like:
    • .fold(): Row-wise reduction by applying a function that reduces two Series to one.
    • .hash_rows(): Hashes each row to a UInt64 value.

• LazyFrame:

  • Does not support row-wise computation methods like .fold() and .hash_rows().

GroupBy Methods Difference

• Key Difference: In DataFrame, you can iterate over groups using .__iter__(), but this is unavailable in LazyFrame.

Exporting Methods Difference

• DataFrame: Can export data to various formats such as Arrow, Numpy, Pandas, dictionaries, and more.
• LazyFrame: Cannot export data since it does not hold data until collected.

Manipulation and Selection Difference

• DataFrame: Offers more methods for manipulation and selection compared to LazyFrame.

Miscellaneous Methods Difference

• LazyFrame: Limited support for miscellaneous methods compared to DataFrame.

Out-of-Core Computation with Lazy API’s Streaming Mode

• Out-of-core computation: Processes data in chunks, allowing computations on datasets too large to fit in RAM.
• Benefit: Shifts the memory limit from the size of RAM to the size of your hard disk, significantly increasing the capacity for handling large datasets.

How to Use Streaming Mode

• Trigger streaming mode by passing streaming=True to:

  • .collect(streaming=True) — collects the result into RAM (final result must fit in RAM).
  • .sink_csv(), .sink_ipc(), .sink_parquet() — writes results to disk.

• How it works:

  • Data is read in chunks of rows.
  • Chunk size depends on the number of available threads and number of columns in the dataset.

Determining Available Threads

• Find the number of logical CPU cores using:

  pl.thread_pool_size()

  • Example output: 12 threads available.

• Adjusting thread count:

  • Modify thread count using environment variables before importing Polars:

  import os
  os.environ["POLARS_MAX_THREADS"] = "2"
  import polars as pl

Chunk Size Calculation in Streaming Mode

• Formula: Chunk size is the smaller of:

  • 1000
  • 50000 / (n_cols * thread_factor)

• Thread factor:

  • 1 if there are 12 or more threads.
  • Greater than 1 if there are fewer threads (larger chunk size for fewer threads).

• Factors affecting chunk size:

  • More threads: Smaller chunk size.
  • More columns: Smaller chunk size (each row uses more RAM).

Adjusting Chunk Size Manually

• If default chunk size causes memory issues, manually set chunk size using:

  pl.Config.set_streaming_chunk_size(1000)

Tips and Tricks for Using Polars

1. Converting Between LazyFrame and DataFrame

• From DataFrame to LazyFrame:

  • Use .lazy() to convert a DataFrame into a LazyFrame. This does not trigger computation, but sets up the data for lazy execution.

  • Example:

    df = pl.DataFrame({"col": [1, 2, 3]})
    lf = df.lazy()

• From LazyFrame to DataFrame:

  • Use .collect() to materialize a LazyFrame, triggering the query execution and storing the result in RAM.

  • Example:

    lf.collect()

2. Joining a DataFrame and LazyFrame

• Error when joining a DataFrame and a LazyFrame directly:

  lf = pl.LazyFrame(
    {"id": [1, 2, 3], "value1": [4, 5, 6]})
  df = pl.DataFrame(
    {"id": [1, 2, 3], "value2": [7, 8, 9]})
  lf.join(df, on="id")  # Results in a TypeError

• Solution:

  • Convert the DataFrame to a LazyFrame using .lazy() or convert the LazyFrame to a DataFrame with .collect().
  • Recommended: Stick with the lazy API for better performance.

  lf.join(df.lazy(), on="id")

3. Caching Intermittent Stages

• To avoid recomputing heavy operations, cache the LazyFrame in memory by chaining .collect().lazy().

  • This collects the result into RAM and returns a LazyFrame that points to the cached data.

• Pattern for optimizing computations:

  lf = pl.LazyFrame(
    {"col1": [1, 2, 3], "col2": [4, 5, 6]})
  
  lf = lf.collect().lazy()  # Cache result in memory
  print(lf.collect())
  
  # Utilizes the cached result
  print(lf.with_columns(pl.col("col1") + 1).collect())

• Benefits: Leverages lazy evaluation while avoiding the need for recomputing resource-intensive stages.