rudolfix
commented
4 months ago @arjun-panchmatia-mechademy we have table_adapter callback that let's you modify the SQL Alchemy Table before it is used for select. that allows to remove certain columns ie. (it is in the sql_database_pipeline.py demo)
def select_columns() -> None:
"""Uses table adapter callback to modify list of columns to be selected"""
pipeline = dlt.pipeline(
pipeline_name="rfam_database",
destination="duckdb",
dataset_name="rfam_data_cols",
full_refresh=True,
)
def table_adapter(table: Table) -> None:
print(table.name)
if table.name == "family":
# this is SqlAlchemy table. _columns are writable
# let's drop updated column
table._columns.remove(table.columns["updated"])
family = sql_table(
credentials="mysql+pymysql://rfamro@mysql-rfam-public.ebi.ac.uk:4497/Rfam",
table="family",
chunk_size=10,
reflection_level="full_with_precision",
table_adapter_callback=table_adapter,
)
# also we do not want the whole table, so we add limit to get just one chunk (10 records)
pipeline.run(family.add_limit(1))
# only 10 rows
print(pipeline.last_trace.last_normalize_info)
# no "updated" column in "family" table
print(pipeline.default_schema.to_pretty_yaml())regarding filters: really good idea! we intended to do it in similar way: just before executing we let user to modify the expression and ie add a WHERE clause
what do you think? those are powerful but advanced interfaces ie. you need to know SQL Alchemy to use them. did you have something simpler in mind?
arjun-panchmatia-mechademy
Source name
sql_database
Describe the data you'd like to see
Background: A fairly common use-case when dealing with clients is having a direct database exposed for extraction. Considering the non-technical nature of these clients, these DBs often have lots of columns that may be irrelevant during the extraction process AND/OR require custom column-wise filtering to make the relevant data available for ingestion. This is best expressed as a SQL query where you can select individual columns and apply filters as needed. Currently, the
sql_databasesource does not have the ability to do level of filtering or column selection. This makes it hard to use directly in pipelines, as often applying a map function still results in a full scan, slowing down critical loads. We often have instances of extreme volumes of data for even 15 minutes of incremental loads and cannot use this source as it would be far too slow, and potentially eat up a lot of in-memory store. We've been working around this by writing custom .sql files and yielding them from a resource, but that can be quite cumbersome from a maintenance perspective and does not allow us to leverage the full breadth of dlt's features.Request: It would be great if we can get some level of control on the data loaded by the sql_database source (or the sql_table) resource. Potentially letting us select our required columns (maybe via a data_selector as is configred in the rest api source) and allow a level of filtering to occur. Aggregation would also be a generally useful function to have but that may be overkill and is probably better suited to post processing / ELT methadology anyway.
Are you a dlt user?
Yes, I'm already a dlt user.
Do you ready to contribute this extension?
No.
dlt destination
filesystem
Additional information
I think just having the ability to select columns and apply filters would go a long way. A big benefit of using ORMs is being DB agnostic. When trying to write an orchestration package, we need to necessarily account for variance and try to be as agnostic as possible, and this would go a long way towards helping us achieve that goal.
Current workaround is as described above: writing raw SQL / using an ORM and then yielding the data from the resource.