From 9cd8bb954685cc072c5af47e227e2c6bb1c91bfa Mon Sep 17 00:00:00 2001 From: LanAnh Date: Tue, 29 Sep 2026 01:48:03 +0700 Subject: [PATCH 1/4] feat(examples): add DuckDB partitioned Parquet export integration recipe --- examples/README.md | 1 + .../integrations/duckdb_parquet_export.py | 86 +++++++++++++++++++ 2 files changed, 87 insertions(+) create mode 100644 examples/integrations/duckdb_parquet_export.py diff --git a/examples/README.md b/examples/README.md index 0e1d85e..007190f 100644 --- a/examples/README.md +++ b/examples/README.md @@ -36,6 +36,7 @@ Located in [`examples/integrations/`](integrations/): | **pandas** | [`pandas_workflow.py`](integrations/pandas_workflow.py) | Drop-in cleaning step within an existing pandas aggregation workflow. | | **Polars** | [`polars_workflow.py`](integrations/polars_workflow.py) | Native `Polars in -> Polars out` execution with downstream Polars query expressions. | | **DuckDB** | [`duckdb_workflow.py`](integrations/duckdb_workflow.py) | Direct cleaning of `DuckDBPyRelation` objects and registering back to DuckDB for SQL. | +| **DuckDB Parquet** | [`duckdb_parquet_export.py`](integrations/duckdb_parquet_export.py) | Reading partitioned Parquet directories, cleaning with FreshData, and exporting back via DuckDB. | | **scikit-learn** | [`sklearn_pipeline.py`](integrations/sklearn_pipeline.py) | Custom `FreshDataCleaner` transformer for native `sklearn.pipeline.Pipeline` usage. | | **Apache Airflow** | [`airflow_task.py`](integrations/airflow_task.py) | Quality gating with `evaluate_trust_gate` and `FreshDataCleanOperator` DAG pattern. | | **Pandera** | [`09_pandera_recipe.py`](09_pandera_recipe.py) | Pre- and post-validation using declarative Pandera DataFrame schemas. | diff --git a/examples/integrations/duckdb_parquet_export.py b/examples/integrations/duckdb_parquet_export.py new file mode 100644 index 0000000..75221ce --- /dev/null +++ b/examples/integrations/duckdb_parquet_export.py @@ -0,0 +1,86 @@ +"""examples/integrations/duckdb_parquet_export.py — Partitioned Parquet Cleaning & Export. + +Problem: +-------- +Analytical data pipelines often store large datasets as partitioned Parquet files +(e.g., partitioned by region or year). Raw partitions often contain messy values, +inconsistent types, and unstandardized strings. FreshData cleans the dataset, and +DuckDB writes the cleaned result back out into a partitioned Parquet destination. + +Installation: +------------- +pip install "freshdata-cleaner[duckdb]" +""" + +from __future__ import annotations + +import tempfile +from pathlib import Path + +import duckdb +import freshdata as fd + + +def main() -> None: + print("=== Integration: DuckDB Partitioned Parquet Export & FreshData ===") + + con = duckdb.connect() + + with tempfile.TemporaryDirectory() as tmp_dir: + base_path = Path(tmp_dir) + raw_parquet_dir = base_path / "raw_sales" + cleaned_parquet_dir = base_path / "cleaned_sales" + + print("\n[1] Creating raw partitioned Parquet dataset...") + con.execute(f""" + CREATE TABLE raw_data AS SELECT * FROM (VALUES + ('TX-101', 'North', ' Acme Corp ', '$15,000.50', '2024-01-10'), + ('TX-102', 'North', 'Beta Ltd', 'N/A', '2024-01-12'), + ('TX-201', 'South', 'Gamma LLC', '$8,400.00', 'invalid_date'), + ('TX-202', 'South', 'Delta Inc', '$22,100.00', '2024-02-01') + ) AS t(transaction_id, region, company_name, revenue, sale_date); + + COPY raw_data TO '{raw_parquet_dir.as_posix()}' ( + FORMAT PARQUET, + PARTITION_BY (region), + OVERWRITE_OR_IGNORE 1 + ); + """) + + print("\n[2] Reading partitioned Parquet dataset into DuckDB relation...") + raw_relation = con.read_parquet( + f"{raw_parquet_dir.as_posix()}/*/*.parquet", + hive_partitioning=True, + ) + print(raw_relation.df()) + + print("\n[3] Cleaning data with FreshData...") + cleaned_df = fd.clean( + raw_relation, + preserve_columns=["transaction_id", "region"], + strategy="conservative", + ) + print(cleaned_df) + + print("\n[4] Exporting cleaned data to partitioned Parquet using DuckDB...") + con.register("cleaned_table", cleaned_df) + + con.execute(f""" + COPY (SELECT * FROM cleaned_table) TO '{cleaned_parquet_dir.as_posix()}' ( + FORMAT PARQUET, + PARTITION_BY (region), + OVERWRITE_OR_IGNORE 1 + ); + """) + + print("\n[5] Verifying exported Parquet files:") + verified_relation = con.read_parquet( + f"{cleaned_parquet_dir.as_posix()}/*/*.parquet", + hive_partitioning=True, + ) + print(verified_relation.df()) + print("\nSuccessfully exported cleaned partitioned dataset!") + + +if __name__ == "__main__": + main() \ No newline at end of file From 71e8c10f07111af0737f8f9a2d0a719997b0ccf2 Mon Sep 17 00:00:00 2001 From: LanAnh Date: Wed, 30 Sep 2026 17:14:40 +0700 Subject: [PATCH 2/4] docs(examples): add docstring and register duckdb parquet export recipe --- examples/integrations/duckdb_parquet_export.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/examples/integrations/duckdb_parquet_export.py b/examples/integrations/duckdb_parquet_export.py index 75221ce..e431144 100644 --- a/examples/integrations/duckdb_parquet_export.py +++ b/examples/integrations/duckdb_parquet_export.py @@ -22,6 +22,15 @@ def main() -> None: + """Execute the end-to-end DuckDB partitioned Parquet cleaning and export workflow. + + Demonstrates: + 1. Creating a sample raw Hive-partitioned Parquet dataset. + 2. Reading the partitioned Parquet files directly into a DuckDB relation. + 3. Cleaning messy values using FreshData. + 4. Registering and re-exporting the cleaned data back into a partitioned Parquet layout. + 5. Validating the integrity of the exported Parquet records. + """ print("=== Integration: DuckDB Partitioned Parquet Export & FreshData ===") con = duckdb.connect() From 3d8607e187f124f273879865512221724bf9ef0e Mon Sep 17 00:00:00 2001 From: LanAnh Date: Fri, 2 Oct 2026 09:47:57 +0700 Subject: [PATCH 3/4] fix(examples): apply ruff linting, context manager, assertions, and materialize relation --- .../integrations/duckdb_parquet_export.py | 130 +++++++++--------- 1 file changed, 65 insertions(+), 65 deletions(-) diff --git a/examples/integrations/duckdb_parquet_export.py b/examples/integrations/duckdb_parquet_export.py index e431144..ece6308 100644 --- a/examples/integrations/duckdb_parquet_export.py +++ b/examples/integrations/duckdb_parquet_export.py @@ -18,77 +18,77 @@ from pathlib import Path import duckdb + import freshdata as fd def main() -> None: - """Execute the end-to-end DuckDB partitioned Parquet cleaning and export workflow. - - Demonstrates: - 1. Creating a sample raw Hive-partitioned Parquet dataset. - 2. Reading the partitioned Parquet files directly into a DuckDB relation. - 3. Cleaning messy values using FreshData. - 4. Registering and re-exporting the cleaned data back into a partitioned Parquet layout. - 5. Validating the integrity of the exported Parquet records. - """ print("=== Integration: DuckDB Partitioned Parquet Export & FreshData ===") - con = duckdb.connect() - - with tempfile.TemporaryDirectory() as tmp_dir: - base_path = Path(tmp_dir) - raw_parquet_dir = base_path / "raw_sales" - cleaned_parquet_dir = base_path / "cleaned_sales" - - print("\n[1] Creating raw partitioned Parquet dataset...") - con.execute(f""" - CREATE TABLE raw_data AS SELECT * FROM (VALUES - ('TX-101', 'North', ' Acme Corp ', '$15,000.50', '2024-01-10'), - ('TX-102', 'North', 'Beta Ltd', 'N/A', '2024-01-12'), - ('TX-201', 'South', 'Gamma LLC', '$8,400.00', 'invalid_date'), - ('TX-202', 'South', 'Delta Inc', '$22,100.00', '2024-02-01') - ) AS t(transaction_id, region, company_name, revenue, sale_date); - - COPY raw_data TO '{raw_parquet_dir.as_posix()}' ( - FORMAT PARQUET, - PARTITION_BY (region), - OVERWRITE_OR_IGNORE 1 - ); - """) - - print("\n[2] Reading partitioned Parquet dataset into DuckDB relation...") - raw_relation = con.read_parquet( - f"{raw_parquet_dir.as_posix()}/*/*.parquet", - hive_partitioning=True, - ) - print(raw_relation.df()) - - print("\n[3] Cleaning data with FreshData...") - cleaned_df = fd.clean( - raw_relation, - preserve_columns=["transaction_id", "region"], - strategy="conservative", - ) - print(cleaned_df) - - print("\n[4] Exporting cleaned data to partitioned Parquet using DuckDB...") - con.register("cleaned_table", cleaned_df) - - con.execute(f""" - COPY (SELECT * FROM cleaned_table) TO '{cleaned_parquet_dir.as_posix()}' ( - FORMAT PARQUET, - PARTITION_BY (region), - OVERWRITE_OR_IGNORE 1 - ); - """) - - print("\n[5] Verifying exported Parquet files:") - verified_relation = con.read_parquet( - f"{cleaned_parquet_dir.as_posix()}/*/*.parquet", - hive_partitioning=True, - ) - print(verified_relation.df()) - print("\nSuccessfully exported cleaned partitioned dataset!") + # 1. Manage DuckDB connection using a context manager + with duckdb.connect() as con: + with tempfile.TemporaryDirectory() as tmp_dir: + base_path = Path(tmp_dir) + raw_parquet_dir = base_path / "raw_sales" + cleaned_parquet_dir = base_path / "cleaned_sales" + + # 2. Setup raw partitioned Parquet dataset + print("\n[1] Creating raw partitioned Parquet dataset...") + con.execute(f""" + CREATE TABLE raw_data AS SELECT * FROM (VALUES + ('TX-101', 'North', ' Acme Corp ', '$15,000.50', '2024-01-10'), + ('TX-102', 'North', 'Beta Ltd', 'N/A', '2024-01-12'), + ('TX-201', 'South', 'Gamma LLC', '$8,400.00', 'invalid_date'), + ('TX-202', 'South', 'Delta Inc', '$22,100.00', '2024-02-01') + ) AS t(transaction_id, region, company_name, revenue, sale_date); + + COPY raw_data TO '{raw_parquet_dir.as_posix()}' ( + FORMAT PARQUET, + PARTITION_BY (region), + OVERWRITE_OR_IGNORE 1 + ); + """) + + # 3. Read partitioned Parquet files directly into a DuckDB relation using glob pattern + print("\n[2] Reading partitioned Parquet dataset into DuckDB relation...") + raw_relation = con.read_parquet( + f"{raw_parquet_dir.as_posix()}/**/*.parquet", + hive_partitioning=True, + ) + print(raw_relation.df()) + + # 4. Clean messy dataset using FreshData (materialize relation via fetchdf() to avoid catalog isolation issues) + print("\n[3] Cleaning data with FreshData...") + cleaned_df = fd.clean( + raw_relation.fetchdf(), + preserve_columns=["transaction_id", "region"], + strategy="conservative", + ) + print(cleaned_df) + + # 5. Register and re-export cleaned data back into partitioned Parquet destination + print("\n[4] Exporting cleaned data to partitioned Parquet using DuckDB...") + con.register("cleaned_table", cleaned_df) + con.execute(f""" + COPY (SELECT * FROM cleaned_table) TO '{cleaned_parquet_dir.as_posix()}' ( + FORMAT PARQUET, + PARTITION_BY (region), + OVERWRITE_OR_IGNORE 1 + ); + """) + + # 6. Verify row count and partition existence for self-validation + print("\n[5] Verifying exported Parquet files:") + verified_relation = con.read_parquet( + f"{cleaned_parquet_dir.as_posix()}/**/*.parquet", + hive_partitioning=True, + ) + print(verified_relation.df()) + + assert len(verified_relation) == 4 + assert (cleaned_parquet_dir / "region=North").exists() + + print("\nSuccessfully exported cleaned partitioned dataset!") if __name__ == "__main__": From 6942e57957749a53e795c47f4c9efa4fb40ad171 Mon Sep 17 00:00:00 2001 From: LanAnh Date: Fri, 2 Oct 2026 09:56:51 +0700 Subject: [PATCH 4/4] fix(examples): escape single quotes in COPY destinations --- examples/integrations/duckdb_parquet_export.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/examples/integrations/duckdb_parquet_export.py b/examples/integrations/duckdb_parquet_export.py index ece6308..66b7a25 100644 --- a/examples/integrations/duckdb_parquet_export.py +++ b/examples/integrations/duckdb_parquet_export.py @@ -32,6 +32,10 @@ def main() -> None: raw_parquet_dir = base_path / "raw_sales" cleaned_parquet_dir = base_path / "cleaned_sales" + # Escape single quotes for SQL literal interpolation + raw_parquet_path = raw_parquet_dir.as_posix().replace("'", "''") + cleaned_parquet_path = cleaned_parquet_dir.as_posix().replace("'", "''") + # 2. Setup raw partitioned Parquet dataset print("\n[1] Creating raw partitioned Parquet dataset...") con.execute(f""" @@ -42,7 +46,7 @@ def main() -> None: ('TX-202', 'South', 'Delta Inc', '$22,100.00', '2024-02-01') ) AS t(transaction_id, region, company_name, revenue, sale_date); - COPY raw_data TO '{raw_parquet_dir.as_posix()}' ( + COPY raw_data TO '{raw_parquet_path}' ( FORMAT PARQUET, PARTITION_BY (region), OVERWRITE_OR_IGNORE 1 @@ -70,7 +74,7 @@ def main() -> None: print("\n[4] Exporting cleaned data to partitioned Parquet using DuckDB...") con.register("cleaned_table", cleaned_df) con.execute(f""" - COPY (SELECT * FROM cleaned_table) TO '{cleaned_parquet_dir.as_posix()}' ( + COPY (SELECT * FROM cleaned_table) TO '{cleaned_parquet_path}' ( FORMAT PARQUET, PARTITION_BY (region), OVERWRITE_OR_IGNORE 1