diff --git a/data_check.py b/data_check.py index e893882..004fdc6 100644 --- a/data_check.py +++ b/data_check.py @@ -1,17 +1,18 @@ +import duckdb +import os import polars as pl -df = pl.read_parquet('data/warehouse/transactions.parquet') -print("Data loaded from Parquet file:") -print(df) -relevant_data = df.sql(''' +db = duckdb.connect('/mnt/bulk/data/ynab_data.duckdb') + + # SELECT * FROM gold.transactions +relevant_data = db.sql(''' SELECT transaction_date, sum(transaction_amount) as total - FROM self + FROM gold.transactions GROUP BY transaction_date ORDER BY transaction_date DESC - ''' -) +''') print("Data after SQL query:") print(relevant_data) diff --git a/main.py b/main.py index ea7011d..f9a0dce 100644 --- a/main.py +++ b/main.py @@ -72,5 +72,3 @@ if __name__ == '__main__': else: logger.error(f'Program exited with code {exit_code}') raise - -# test comment diff --git a/pipeline/duckdb_layer.py b/pipeline/duckdb_layer.py new file mode 100644 index 0000000..f1c822b --- /dev/null +++ b/pipeline/duckdb_layer.py @@ -0,0 +1,241 @@ +"""DuckDB-based silver and gold layer for the YNAB data pipeline.""" + +import os +from datetime import date, timedelta + +import duckdb + + +class DuckDBLayer: + """Orchestrates silver (external base) and gold (views + materialized) layers via DuckDB.""" + + def __init__(self, base_path: str, warehouse_path: str, logger): + self.base_path = base_path + self.warehouse_path = warehouse_path + self.logger = logger + self.db = None + os.makedirs(self.warehouse_path, exist_ok=True) + self._init() + + def _silver_parquet_expr(self, entity: str) -> str: + return f"read_parquet('{self.base_path}/{entity}.parquet')" + + def _init(self): + self.db = duckdb.connect('/mnt/bulk/data/ynab_data.duckdb') + self._setup_schemas() + self._create_silver_external_tables() + self._drop_gold() + self._create_gold_views() + self._create_gold_dates() + self._create_gold_facts() + + def _setup_schemas(self): + self.db.execute("CREATE SCHEMA IF NOT EXISTS silver") + self.db.execute("CREATE SCHEMA IF NOT EXISTS gold") + + def _create_silver_external_tables(self): + entities = ['accounts', 'categories', 'payees', 'transactions', 'scheduled_transactions'] + for entity in entities: + sql = f"CREATE TABLE IF NOT EXISTS silver.{entity} AS SELECT * FROM {self._silver_parquet_expr(entity)}" + try: + self.db.execute(sql) + self.logger.info(f"Created silver.{entity}") + except Exception as e: + self.logger.error(f"Failed to create silver.{entity}: {e}") + self.logger.info("Silver layer initialized") + + def _drop_gold(self): + for name in ['accounts', 'categories', 'payees']: + self.db.execute(f"DROP VIEW IF EXISTS gold.{name}") + self.db.execute("DROP TABLE IF EXISTS gold.dates") + self.db.execute("DROP TABLE IF EXISTS gold.transactions") + self.db.execute("DROP TABLE IF EXISTS gold.scheduled_transactions") + + # ── gold views (dimensions) ───── + + def _create_gold_accounts_view(self): + self.logger.info("Transforming and creating gold.accounts view") + sql = """ + SELECT + id AS account_id, + name AS account_name, + "type" AS account_type, + on_budget, + closed, + COALESCE(CAST(note AS VARCHAR), 'none') AS note, + balance / 1000.0 AS balance, + cleared_balance / 1000.0 AS cleared_balance, + uncleared_balance / 1000.0 AS uncleared_balance, + deleted + FROM silver.accounts + """ + try: + self.db.execute(f"CREATE VIEW gold.accounts AS {sql}") + self.logger.info("Created gold.accounts view") + except Exception as e: + self.logger.error(f"Failed to create gold.accounts view: {e}") + + def _create_gold_categories_view(self): + self.logger.info("Transforming and creating gold.categories view") + sql = """ + SELECT + id AS category_id, + name AS category_name, + category_group_name, + hidden, + COALESCE(CAST(note AS VARCHAR), 'none') AS note, + budgeted / 1000.0 AS budgeted, + activity / 1000.0 AS activity, + balance / 1000.0 AS balance, + deleted + FROM silver.categories + """ + try: + self.db.execute(f"CREATE VIEW gold.categories AS {sql}") + self.logger.info("Created gold.categories view") + except Exception as e: + self.logger.error(f"Failed to create gold.categories view: {e}") + + def _create_gold_payees_view(self): + self.logger.info("Transforming and creating gold.payees view") + sql = """ + SELECT + id AS payee_id, + name AS payee_name, + deleted + FROM silver.payees + """ + try: + self.db.execute(f"CREATE VIEW gold.payees AS {sql}") + self.logger.info("Created gold.payees view") + except Exception as e: + self.logger.error(f"Failed to create gold.payees view: {e}") + + def _create_gold_views(self): + self._create_gold_accounts_view() + self._create_gold_categories_view() + self._create_gold_payees_view() + + # ── gold dates (materialized as Parquet, loaded into DuckDB) ───── + + def _create_gold_dates(self): + dates_path = os.path.join(self.warehouse_path, 'dates.parquet') + self.logger.info("Creating gold.dates dimension") + try: + start = date(2020, 1, 1) + end = date(2030, 12, 31) + days = (end - start).days + 1 + rows = [] + for i in range(days): + d = start + timedelta(days=i) + month_z = str(d.month).zfill(2) + day_z = str(d.day).zfill(2) + rows.append({ + 'date_id': f"{d.year}{month_z}{day_z}", + 'date': d, + 'year': int(d.year), + 'month': int(d.month), + 'day': int(d.day), + 'weekday': int(d.isoweekday()), + 'is_weekday': 1 if d.isoweekday() < 6 else 0, + }) + + import pyarrow as pa + import pyarrow.parquet as pq + dates_df = pa.Table.from_pylist(rows) + dates_df = dates_df.cast(pa.schema([ + ('date_id', pa.string()), + ('date', pa.date32()), + ('year', pa.int32()), + ('month', pa.int8()), + ('day', pa.int8()), + ('weekday', pa.int8()), + ('is_weekday', pa.int8()), + ])) + pq.write_table(dates_df, dates_path) + + self.db.execute(f"CREATE TABLE gold.dates AS SELECT * FROM read_parquet('{dates_path}')") + self.logger.info(f"Created gold.dates and persisted to {dates_path}") + except Exception as e: + self.logger.error(f"Failed to create gold.dates: {e}") + + # ── gold facts (materialized as Parquet) ───── + + def _create_gold_transactions_fact(self): + transactions_path = os.path.join(self.warehouse_path, 'transactions.parquet') + self.logger.info("Transforming and persisting gold.transactions fact") + sql = f""" + COPY ( + SELECT + memo, + cleared, + approved, + COALESCE(CAST(flag_color AS VARCHAR), 'none') AS flag_color, + account_id, + payee_id, + category_id, + transfer_account_id, + id AS transaction_id, + CAST(YEAR(CAST(date AS DATE)) AS VARCHAR) + || LPAD(CAST(MONTH(CAST(date AS DATE)) AS VARCHAR), 2, '0') + || LPAD(CAST(DAY(CAST(date AS DATE)) AS VARCHAR), 2, '0') AS transaction_date, + amount / 1000.0 AS transaction_amount, + deleted + FROM silver.transactions + ) TO '{transactions_path}' (FORMAT PARQUET) + """ + try: + self.db.execute(sql) + self.logger.info(f"Persisted transactions.fact to {transactions_path}") + except Exception as e: + self.logger.error(f"Failed to create transactions.fact: {e}") + + def _create_gold_scheduled_transactions_fact(self): + st_path = os.path.join(self.warehouse_path, 'scheduled_transactions.parquet') + self.logger.info("Transforming and persisting gold.scheduled_transactions fact") + sql = f""" + COPY ( + SELECT + CAST(date_first AS DATE) AS date_first, + CAST(date_next AS DATE) AS date_next, + frequency, + COALESCE(memo, 'none') AS memo, + COALESCE(CAST(flag_color AS VARCHAR), 'none') AS flag_color, + account_id, + payee_id, + category_id, + transfer_account_id, + id AS scheduled_transaction_id, + amount / 1000.0 AS scheduled_transaction_amount + FROM silver.scheduled_transactions + ) TO '{st_path}' (FORMAT PARQUET) + """ + try: + self.db.execute(sql) + self.logger.info(f"Persisted scheduled_transactions.fact to {st_path}") + except Exception as e: + self.logger.error(f"Failed to create scheduled_transactions.fact: {e}") + + def _create_gold_facts(self): + self._create_gold_transactions_fact() + self._create_gold_scheduled_transactions_fact() + # Create DuckDB tables from the persistent Parquet files so they're queryable + self._register_facts() + + def _register_facts(self): + """Register the fact Parquet files as DuckDB tables.""" + try: + self.db.execute(f"CREATE TABLE gold.transactions AS SELECT * FROM read_parquet('{os.path.join(self.warehouse_path, 'transactions.parquet')}')") + self.db.execute(f"CREATE TABLE gold.scheduled_transactions AS SELECT * FROM read_parquet('{os.path.join(self.warehouse_path, 'scheduled_transactions.parquet')}')") + self.logger.info("Registered fact tables in DuckDB") + except Exception as e: + self.logger.error(f"Failed to register fact tables: {e}") + + def close(self): + if self.db: + self.db.close() + + +def get_duckdb_layer(base_path: str, warehouse_path: str, logger) -> DuckDBLayer: + """Factory function to create a DuckDB layer and perform all transformations.""" + return DuckDBLayer(base_path, warehouse_path, logger) diff --git a/pipeline/pipeline_main.py b/pipeline/pipeline_main.py index 09adaad..44a692f 100644 --- a/pipeline/pipeline_main.py +++ b/pipeline/pipeline_main.py @@ -1,19 +1,19 @@ '''Module to run the data pipeline''' -from pipeline import dimensions, facts, ingest, raw_to_base +from pipeline import duckdb_layer, ingest, raw_to_base def pipeline_main(config, logger): '''Run the data pipeline''' logger.info('Starting data pipeline') - ingest.Ingest(config,logger).start_ingestion() - raw_to_base.RawToBase(config,logger) - dimensions.DimAccounts(config,logger) - dimensions.DimCategories(config,logger) - dimensions.DimPayees(config,logger) - dimensions.DimDate(config,logger) - facts.FactTransactions(config,logger) - facts.FactScheduledTransactions(config,logger) + ingest.Ingest(config, logger).start_ingestion() + raw_to_base.RawToBase(config, logger) + + duckdb_layer.get_duckdb_layer( + config['base_data_path'], + config['warehouse_data_path'], + logger + ) logger.info('Data pipeline completed successfully') diff --git a/pipeline/raw_to_base.py b/pipeline/raw_to_base.py index e7d880d..1f3cbe1 100644 --- a/pipeline/raw_to_base.py +++ b/pipeline/raw_to_base.py @@ -8,7 +8,6 @@ import polars as pl import config.exit_codes as ec -#test comment for pr check class RawToBase: def __init__(self, config: dict[str, Any],logger): diff --git a/pyproject.toml b/pyproject.toml index 5d670e8..953ca41 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,6 +5,7 @@ description = "Add your description here" readme = "README.md" requires-python = ">=3.13" dependencies = [ + "duckdb>=1.5.5", "pandas>=3.0.5", "polars>=1.43.0", "pyarrow>=25.0.0", diff --git a/uv.lock b/uv.lock index a5f8ea3..8bb5041 100644 --- a/uv.lock +++ b/uv.lock @@ -81,6 +81,7 @@ name = "data-pipeline-for-ynab" version = "0.1.0" source = { virtual = "." } dependencies = [ + { name = "duckdb" }, { name = "pandas" }, { name = "polars" }, { name = "pyarrow" }, @@ -93,6 +94,7 @@ dependencies = [ [package.metadata] requires-dist = [ + { name = "duckdb", specifier = ">=1.5.5" }, { name = "pandas", specifier = ">=3.0.5" }, { name = "polars", specifier = ">=1.43.0" }, { name = "pyarrow", specifier = ">=25.0.0" }, @@ -103,6 +105,28 @@ requires-dist = [ { name = "ruff", specifier = ">=0.16.0" }, ] +[[package]] +name = "duckdb" +version = "1.5.5" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/7d/19/e57151753576373c6696a12022648546cca6038e8833fda2908ee2342d9b/duckdb-1.5.5.tar.gz", hash = "sha256:72f33ee57ca7595b23957671a2cc7f7fe2be0ecc2d68f63abedcfcaa3a5c1238", size = 18066741, upload-time = "2026-07-22T10:55:17.819Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/47/37/4a38116e7700720fd152c666292214fd3abdf916496991296d8d1f66efbf/duckdb-1.5.5-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:cd98829b67788609017e65c761bd42a5dd0f9129441bed8bda4d6881ccf819f0", size = 32754294, upload-time = "2026-07-22T10:54:29.822Z" }, + { url = "https://files.pythonhosted.org/packages/66/42/7d392f1ba1eee0eaf4ab4c8c7a604bfe3536cd63f979cf5c98798664f807/duckdb-1.5.5-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:feead93c56679b79592d437c62975d39cb67adedffa7592c763baf8160ac7366", size = 17368211, upload-time = "2026-07-22T10:54:33.359Z" }, + { url = "https://files.pythonhosted.org/packages/9f/a5/0a6f4fa60562faa615e55e15bd1953a2f2b17a8edd8105e5cda215e43457/duckdb-1.5.5-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:49c963d9469373d7aba8d750d9ea565ab823e94166efed953f184dd9b169b98c", size = 15509136, upload-time = "2026-07-22T10:54:36.369Z" }, + { url = "https://files.pythonhosted.org/packages/e4/cb/023c89f51978545b9fab318581bba0c457a58e7530d2d933e54ae7d8647c/duckdb-1.5.5-cp313-cp313-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:a736217825461732b5442d05a220f3da2e23a0dae114efbf08c9bf171b53098a", size = 19392147, upload-time = "2026-07-22T10:54:39.551Z" }, + { url = "https://files.pythonhosted.org/packages/3e/c5/41bef391fb8b23dbc133c9f2ba016e7a7a8124513d2cc1b430f1897d87e4/duckdb-1.5.5-cp313-cp313-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:078e6a60dd8eedde5832f45422ca5c4a6b8c837aeabd8a56ca0b7d933f588053", size = 21511060, upload-time = "2026-07-22T10:54:42.788Z" }, + { url = "https://files.pythonhosted.org/packages/07/9f/c44dfc1f924ac29b3252dc1b91393c01d009dbfe9f8ed33f10b986151bd1/duckdb-1.5.5-cp313-cp313-win_amd64.whl", hash = "sha256:6826504277dba513c0c5d71d828456c94d729c9d2482f94b2e289f90a9167e28", size = 13168028, upload-time = "2026-07-22T10:54:46.127Z" }, + { url = "https://files.pythonhosted.org/packages/ca/88/591384b2cd59abddd6f5dc175e60374f9abae6064429f0c4402854c10f44/duckdb-1.5.5-cp313-cp313-win_arm64.whl", hash = "sha256:baa9c5702002fabb559ded2a39008f9f421fcbc7237d388b8213eff1e08858de", size = 13989955, upload-time = "2026-07-22T10:54:49.262Z" }, + { url = "https://files.pythonhosted.org/packages/3e/56/12c65bfa2d2605b81981b264788891bcf11ec72227889554cead5d8d13b9/duckdb-1.5.5-cp314-cp314-macosx_10_15_universal2.whl", hash = "sha256:8e6413dd40facb7b8ab21bd844450cd8f549b29e138635be9cf090ef4d2049e2", size = 32761946, upload-time = "2026-07-22T10:54:53.412Z" }, + { url = "https://files.pythonhosted.org/packages/b9/46/682ce155f17e0d2822d4f13ee3db9ca4b5b7c2da61b841b2629035e1f4bc/duckdb-1.5.5-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:64078acfd16541132ac6e191eb81b2845554444a0305cc1aa581ba107e514aa8", size = 17375069, upload-time = "2026-07-22T10:54:57.269Z" }, + { url = "https://files.pythonhosted.org/packages/39/ce/a24bcbd3289c8f305a430759c5fc12242740b4af3e17f7593f3a34e333d2/duckdb-1.5.5-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:8c11775cc99a447618d5f1840126db17f2652f3eae05529df4f81f40e2df7151", size = 15519791, upload-time = "2026-07-22T10:55:00.681Z" }, + { url = "https://files.pythonhosted.org/packages/d9/76/3a01afbc615c1d418c0de58a6b68ac5ce2a8563232c0464bfbc2ce552398/duckdb-1.5.5-cp314-cp314-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:77bbc1e6ba12e1e06f9020117bdf848627ecfdf36f907550e62e008e6109dece", size = 19398251, upload-time = "2026-07-22T10:55:04.168Z" }, + { url = "https://files.pythonhosted.org/packages/a1/43/3a5e81d1728f4d234c79bfe385808ee7c04834f7c37a4b5c257459c25614/duckdb-1.5.5-cp314-cp314-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:fbf0f2d48b43c6c304d00463b463c27ead6c4b01c3c1816b750f728decf71afe", size = 21513851, upload-time = "2026-07-22T10:55:07.864Z" }, + { url = "https://files.pythonhosted.org/packages/91/41/fc7c829172c60ca22485251eab285f4f1a0d87b486a024c726f21471d86e/duckdb-1.5.5-cp314-cp314-win_amd64.whl", hash = "sha256:9dc826c4b50e64f6c4e4d07a3a9cb075ef70ba3899dc43ec5493dc3d7b04b353", size = 13691858, upload-time = "2026-07-22T10:55:11.181Z" }, + { url = "https://files.pythonhosted.org/packages/e1/2c/95d9216b79e9273689d7ebce125a54503ed0c9bd7da931f0265888e99779/duckdb-1.5.5-cp314-cp314-win_arm64.whl", hash = "sha256:63e48d4b74b15aeacd688976432a7225163df8c226eddeb8536bba2d4d4ff433", size = 14470180, upload-time = "2026-07-22T10:55:14.445Z" }, +] + [[package]] name = "idna" version = "3.18"