Tester des pipelines Polars

Mise à l'échelle et optimisation des pipelines de données avec Polars

Liam Brannigan

Data Scientist & Polars Contributor

Pourquoi tester les pipelines ?

Diagramme montrant deux DataFrames qui diffèrent dans une cellule.

Mise à l'échelle et optimisation des pipelines de données avec Polars

Pourquoi tester les pipelines ?

Diagramme montrant une transformation Polars testée par rapport à un DataFrame attendu.

Mise à l'échelle et optimisation des pipelines de données avec Polars

Requêtes complétées par service

def completed_by_department(requests: pl.LazyFrame) -> pl.LazyFrame:
    return
Mise à l'échelle et optimisation des pipelines de données avec Polars

Requêtes complétées par service

def completed_by_department(requests: pl.LazyFrame) -> pl.LazyFrame:
    return (
        requests




    )
Mise à l'échelle et optimisation des pipelines de données avec Polars

Requêtes complétées par service

def completed_by_department(requests: pl.LazyFrame) -> pl.LazyFrame:
    return (
        requests
        .group_by("DEPARTMENT")
        .len()


    )
Mise à l'échelle et optimisation des pipelines de données avec Polars

Requêtes complétées par service

def completed_by_department(requests: pl.LazyFrame) -> pl.LazyFrame:
    return (
        requests
        .group_by("DEPARTMENT")
        .len()
        .with_columns(pl.col("len").cast(pl.Int32))
        .sort("DEPARTMENT")
    )
Mise à l'échelle et optimisation des pipelines de données avec Polars

Données de test

sample = pl.DataFrame({
    "SR_NUMBER": ["SR1", "SR2", "SR3", "SR4"],
    "DEPARTMENT": ["Sanitation", "Water", "Sanitation", "Aviation"],
    "STATUS": ["Completed", "Open", "Completed", "Completed"],
})

shape: (4, 3)
| SR_NUMBER | DEPARTMENT | STATUS    |
| ---       | ---        | ---       |
| str       | str        | str       |
|-----------|------------|-----------|
| SR1       | Sanitation | Completed |
| SR2       | Water      | Open      |
| SR3       | Sanitation | Completed |
| SR4       | Aviation   | Completed |
Mise à l'échelle et optimisation des pipelines de données avec Polars

Résultat attendu

expected = pl.DataFrame(
    {
        "DEPARTMENT": ["Aviation", "Sanitation"],
        "len": [1, 2],
    }
)

shape: (2, 2)
| DEPARTMENT | len |
| ---        | --- |
| str        | i64 |
|------------|-----|
| Aviation   | 1   |
| Sanitation | 2   |
Mise à l'échelle et optimisation des pipelines de données avec Polars

Obtenir le résultat réel

actual = completed_by_department(
Mise à l'échelle et optimisation des pipelines de données avec Polars

Obtenir le résultat réel

actual = completed_by_department(
    sample.lazy()
).collect()
Mise à l'échelle et optimisation des pipelines de données avec Polars

Obtenir le résultat réel

actual = completed_by_department(
    sample.lazy()
).collect()

shape: (3, 2)
| DEPARTMENT | len |
| ---        | --- |
| str        | i32 |
|------------|-----|
| Aviation   | 1   |
| Sanitation | 2   |
| Water      | 1   |
Mise à l'échelle et optimisation des pipelines de données avec Polars

Comparer avec equals

actual.equals(expected)

False
Mise à l'échelle et optimisation des pipelines de données avec Polars

Importer les tests Polars

from polars.testing import (
    assert_frame_equal,
    assert_schema_equal,
)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Tester le schéma

assert_schema_equal(


)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Tester le schéma

assert_schema_equal(
    actual.schema,
    expected.schema,
)

AssertionError: Schemas are different (dtypes do not match)
[left]: [String, Int32]
[right]: [String, Int64]
Mise à l'échelle et optimisation des pipelines de données avec Polars

Corriger le schéma attendu

expected = pl.DataFrame(
    {
        "DEPARTMENT": ["Aviation", "Sanitation"], "len": [1, 2],
    },
    schema={"DEPARTMENT": pl.String, "len": pl.Int32},
)

shape: (2, 2)
| DEPARTMENT | len |
| ---        | --- |
| str        | i32 |
|------------|-----|
| Aviation   | 1   |
| Sanitation | 2   |
Mise à l'échelle et optimisation des pipelines de données avec Polars

Tester le schéma

assert_schema_equal(
    actual.schema,
    expected.schema,
)
print("Schema test passed!")

Schema test passed!
Mise à l'échelle et optimisation des pipelines de données avec Polars

Tester le cadre de données

assert_frame_equal(


)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Tester le cadre de données

assert_frame_equal(
    actual,
    expected,
)

AssertionError: DataFrames are different height (row count) mismatch
[left]: 3
[right]: 2
Mise à l'échelle et optimisation des pipelines de données avec Polars

Comparer le réel et l'attendu

shape: (3, 2)
| DEPARTMENT | len |
| ---        | --- |
| str        | i32 |
|------------|-----|
| Aviation   | 1   |
| Sanitation | 2   |
| Water      | 1   |

shape: (2, 2)
| DEPARTMENT | len |
| ---        | --- |
| str        | i32 |
|------------|-----|
| Aviation   | 1   |
| Sanitation | 2   |
Mise à l'échelle et optimisation des pipelines de données avec Polars

Corriger la requête

def completed_by_department(requests: pl.LazyFrame) -> pl.LazyFrame:
    return (
        requests
        .filter(pl.col("STATUS") == "Completed")
        .group_by("DEPARTMENT")
        .len()
        .with_columns(pl.col("len").cast(pl.Int32))
        .sort("DEPARTMENT")
    )

actual = completed_by_department(
    sample.lazy()
).collect()
Mise à l'échelle et optimisation des pipelines de données avec Polars

Assertions finales

assert_schema_equal(actual.schema, expected.schema)
assert_frame_equal(actual, expected)
print("All tests passed!")

All tests passed!
Mise à l'échelle et optimisation des pipelines de données avec Polars

Passons à la pratique !

Mise à l'échelle et optimisation des pipelines de données avec Polars

Preparing Video For Download...