Add dlt-based ingestion app (app_etl_dlt)#4
Conversation
Alternative to app_etl's hand-rolled Postgres->GCS->BigQuery pipeline, letting dlt own the chunked extraction, Parquet staging, and BigQuery load-job plumbing. Config-driven: one TableConfig entry per table (tables.py) dispatches to a generic incremental or snapshot path (runner.py), rather than a new job file per table. Verified end-to-end against real infrastructure: payment_v2.payments (Cloud SQL, IAM-authenticated via Cloud SQL Auth Proxy) -> GCS Parquet staging -> BigQuery, including the jsonb-column handling (BigQuery can't load dlt's json data_type from Parquet, worked around via has_json_columns + autodetect_schema). Also fixes a latent pytest module collision (both app_etl and app_etl_dlt have tests/test_scaffold.py) by switching to --import-mode=importlib in the root pyproject.toml.
|
|
||
| def _pipeline(table_name: str, settings: Settings) -> dlt.Pipeline: | ||
| return dlt.pipeline( | ||
| pipeline_name=f"ingest_{table_name}", |
There was a problem hiding this comment.
One thing i was considering is whether we should group the ingestion of multiple tables in one pipeline. The reason is mainly to reduce the number of cloud run jobs we will create and maintain. I am thinking a natural split is to group the sources from the same DB because the pipeline can not connect to multiple DBs at once. So for example we create a single pipeline for all tables we want to ingest from payment_v2 DB and so on
| destination=dlt.destinations.bigquery( | ||
| project_id=settings.gcp_project, location=settings.bq_location | ||
| ), | ||
| staging=dlt.destinations.filesystem(bucket_url=f"gs://{settings.gcs_bucket}"), |
There was a problem hiding this comment.
we can add sub folders in GCS to avoid having all parquet files in a single folder, e.g. f"gs://{settings.gcs_bucket}/{settings.bq_dataset_raw}"
| resource, | ||
| table_name=table.name, | ||
| write_disposition="replace", | ||
| loader_file_format="parquet", |
There was a problem hiding this comment.
there is an additional argument refresh we can use to allow DLT to fully replace the data in case we want a clean ingestion again. We can control that from env vars
|
Thanks Aleena for the PR! I think this can immediately go into app_etl directly and we can remove all the hand-rolled implementation! |
2 participants
Alternative to app_etl's hand-rolled Postgres->GCS->BigQuery pipeline, letting dlt own the chunked extraction, Parquet staging, and BigQuery load-job plumbing. Config-driven: one TableConfig entry per table (tables.py) dispatches to a generic incremental or snapshot path (runner.py), rather than a new job file per table.
Verified end-to-end against real infrastructure: payment_v2.payments (Cloud SQL, IAM-authenticated via Cloud SQL Auth Proxy) -> GCS Parquet staging -> BigQuery, including the jsonb-column handling (BigQuery can't load dlt's json data_type from Parquet, worked around via has_json_columns + autodetect_schema).
Also fixes a latent pytest module collision (both app_etl and app_etl_dlt have tests/test_scaffold.py) by switching to --import-mode=importlib in the root pyproject.toml.