Testování Polars pipeline
Scaling and Optimizing Data Pipelines with Polars
Liam Brannigan
Data Scientist & Polars Contributor
Proč testovat pipeline?
Proč testovat pipeline?
Dokončené požadavky podle oddělení
def completed_by_department(requests: pl.LazyFrame) -> pl.LazyFrame:
return
Dokončené požadavky podle oddělení
def completed_by_department(requests: pl.LazyFrame) -> pl.LazyFrame:
return (
requests
)
Dokončené požadavky podle oddělení
def completed_by_department(requests: pl.LazyFrame) -> pl.LazyFrame:
return (
requests
.group_by("DEPARTMENT")
.len()
)
Dokončené požadavky podle oddělení
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")
)
Vstupní data testu
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 |
Očekávaný výstup
expected = pl.DataFrame(
{
"DEPARTMENT": ["Aviation", "Sanitation"],
"len": [1, 2],
}
)
shape: (2, 2)
| DEPARTMENT | len |
| --- | --- |
| str | i64 |
|------------|-----|
| Aviation | 1 |
| Sanitation | 2 |
Získání skutečného výsledku
actual = completed_by_department(
Získání skutečného výsledku
actual = completed_by_department(
sample.lazy()
).collect()
Získání skutečného výsledku
actual = completed_by_department(
sample.lazy()
).collect()
shape: (3, 2)
| DEPARTMENT | len |
| --- | --- |
| str | i32 |
|------------|-----|
| Aviation | 1 |
| Sanitation | 2 |
| Water | 1 |
Porovnání pomocí equals
actual.equals(expected)
False
Import testovacího modulu Polars
from polars.testing import (
assert_frame_equal,
assert_schema_equal,
)
Testování schématu
assert_schema_equal(
)
Testování schématu
assert_schema_equal(
actual.schema,
expected.schema,
)
AssertionError: Schemas are different (dtypes do not match)
[left]: [String, Int32]
[right]: [String, Int64]
Oprava očekávaného schématu
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 |
Testování schématu
assert_schema_equal(
actual.schema,
expected.schema,
)
print("Schema test passed!")
Schema test passed!
Testování DataFrame
assert_frame_equal(
)
Testování DataFrame
assert_frame_equal(
actual,
expected,
)
AssertionError: DataFrames are different height (row count) mismatch
[left]: 3
[right]: 2
Porovnání skutečného a očekávaného výsledku
shape: (3, 2)
| DEPARTMENT | len |
| --- | --- |
| str | i32 |
|------------|-----|
| Aviation | 1 |
| Sanitation | 2 |
| Water | 1 |
shape: (2, 2)
| DEPARTMENT | len |
| --- | --- |
| str | i32 |
|------------|-----|
| Aviation | 1 |
| Sanitation | 2 |
Oprava dotazu
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()
Závěrečná ověření
assert_schema_equal(actual.schema, expected.schema)
assert_frame_equal(actual, expected)
print("All tests passed!")
All tests passed!
Pojďme si procvičit!
Scaling and Optimizing Data Pipelines with Polars
