Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

prepare dataset release & docs updates #2126

Open
wants to merge 9 commits into
base: devel
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions dlt/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,6 @@
)
from dlt.pipeline import progress
from dlt import destinations
from dlt.destinations.dataset import dataset as _dataset

pipeline = _pipeline
current = _current
Expand Down Expand Up @@ -80,7 +79,6 @@
"TCredentials",
"sources",
"destinations",
"_dataset",
]

# verify that no injection context was created
Expand Down
4 changes: 4 additions & 0 deletions dlt/common/destination/reference.py
Original file line number Diff line number Diff line change
Expand Up @@ -592,6 +592,10 @@ def __getattr__(self, table: str) -> SupportsReadableRelation: ...

def ibis(self) -> IbisBackend: ...

def row_counts(
self, *, data_tables: bool = True, dlt_tables: bool = False, table_names: List[str] = None
) -> SupportsReadableRelation: ...


class JobClientBase(ABC):
def __init__(
Expand Down
28 changes: 27 additions & 1 deletion dlt/destinations/dataset/dataset.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Any, Union, TYPE_CHECKING
from typing import Any, Union, TYPE_CHECKING, List

from dlt.common.json import json

Expand Down Expand Up @@ -133,6 +133,32 @@ def table(self, table_name: str) -> SupportsReadableRelation:
table_name=table_name,
) # type: ignore[abstract]

def row_counts(
self, *, data_tables: bool = True, dlt_tables: bool = False, table_names: List[str] = None
) -> SupportsReadableRelation:
"""Returns a dictionary of table names and their row counts, returns counts of all data tables by default"""
"""If table_names is provided, only the tables in the list are returned regardless of the data_tables and dlt_tables flags"""

selected_tables = table_names or []
if not selected_tables:
if data_tables:
selected_tables += self.schema.data_table_names(seen_data_only=True)
if dlt_tables:
selected_tables += self.schema.dlt_table_names()

# Build UNION ALL query to get row counts for all selected tables
queries = []
for table in selected_tables:
queries.append(
f"SELECT '{table}' as table_name, COUNT(*) as row_count FROM"
f" {self.sql_client.make_qualified_table_name(table)}"
)

query = " UNION ALL ".join(queries)

# Execute query and build result dict
return self(query)

def __getitem__(self, table_name: str) -> SupportsReadableRelation:
"""access of table via dict notation"""
return self.table(table_name)
Expand Down
2 changes: 1 addition & 1 deletion dlt/pipeline/pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -1750,7 +1750,7 @@ def __getstate__(self) -> Any:
# pickle only the SupportsPipeline protocol fields
return {"pipeline_name": self.pipeline_name}

def _dataset(
def dataset(
self, schema: Union[Schema, str, None] = None, dataset_type: TDatasetType = "auto"
) -> SupportsReadableDataset:
"""Returns a dataset object for querying the destination data.
Expand Down
27 changes: 17 additions & 10 deletions docs/website/docs/build-a-pipeline-tutorial.md
Original file line number Diff line number Diff line change
Expand Up @@ -262,20 +262,30 @@ In this example, the first pipeline loads the data using `pipedrive_source()`. T

#### [Using the `dlt` SQL client](dlt-ecosystem/transformations/sql.md)

Another option is to leverage the `dlt` SQL client to query the loaded data and perform transformations using SQL statements. You can execute SQL statements that change the database schema or manipulate data within tables. Here's an example of inserting a row into the `customers` table using the `dlt` SQL client:
Another option is to leverage the `dlt` SQL client to query the loaded data and perform transformations using SQL statements. You can execute SQL statements that change the database schema or manipulate data within tables. Here's an example of creating a new table with aggregated sales data in duckdb:

```py
pipeline = dlt.pipeline(destination="bigquery", dataset_name="crm")
pipeline = dlt.pipeline(destination="duckdb", dataset_name="crm")

with pipeline.sql_client() as client:
client.execute_sql(
"INSERT INTO customers VALUES (%s, %s, %s)", 10, "Fred", "[email protected]"
)
""" CREATE TABLE aggregated_sales AS
SELECT
category,
region,
SUM(amount) AS total_sales,
AVG(amount) AS average_sales
FROM
sales
GROUP BY
category,
region;
""")
```

In this example, the `execute_sql` method of the SQL client allows you to execute SQL statements. The statement inserts a row with values into the `customers` table.

#### [Using Pandas](dlt-ecosystem/transformations/pandas.md)
#### [Using Pandas](dlt-ecosystem/transformations/python.md)

You can fetch query results as Pandas data frames and perform transformations using Pandas functionalities. Here's an example of reading data from the `issues` table in DuckDB and counting reaction types using Pandas:

Expand All @@ -287,11 +297,8 @@ pipeline = dlt.pipeline(
dev_mode=True
)

with pipeline.sql_client() as client:
with client.execute_query(
'SELECT "reactions__+1", "reactions__-1", reactions__laugh, reactions__hooray, reactions__rocket FROM issues'
) as cursor:
reactions = cursor.df()
# get a dataframe of all reactions from the dataset
reactions = pipeline.dataset().issues.select("reactions__+1", "reactions__-1", "reactions__laugh", "reactions__hooray", "reactions__rocket").df()

counts = reactions.sum(0).sort_values(0, ascending=False)
```
Expand Down
2 changes: 1 addition & 1 deletion docs/website/docs/dlt-ecosystem/destinations/duckdb.md
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ to disable tz adjustments.

## Destination configuration

By default, a DuckDB database will be created in the current working directory with a name `<pipeline_name>.duckdb` (`chess.duckdb` in the example above). After loading, it is available in `read/write` mode via `with pipeline.sql_client() as con:`, which is a wrapper over `DuckDBPyConnection`. See [duckdb docs](https://duckdb.org/docs/api/python/overview#persistent-storage) for details.
By default, a DuckDB database will be created in the current working directory with a name `<pipeline_name>.duckdb` (`chess.duckdb` in the example above). After loading, it is available in `read/write` mode via `with pipeline.sql_client() as con:`, which is a wrapper over `DuckDBPyConnection`. See [duckdb docs](https://duckdb.org/docs/api/python/overview#persistent-storage) for details. If you want to read data, use [datasets](../../general-usage/dataset-access/dataset) instead of the sql client.

The `duckdb` credentials do not require any secret values. [You are free to pass the credentials and configuration explicitly](../../general-usage/destination.md#pass-explicit-credentials). For example:
```py
Expand Down
8 changes: 4 additions & 4 deletions docs/website/docs/dlt-ecosystem/transformations/dbt/dbt.md
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
---
title: Transform the data with dbt
title: Transforming data with dbt
description: Transforming the data loaded by a dlt pipeline with dbt
keywords: [transform, dbt, runner]
---

# Transform the data with dbt
# Transforming data with dbt

[dbt](https://github.com/dbt-labs/dbt-core) is a framework that allows for the simple structuring of your transformations into DAGs. The benefits of using dbt include:

Expand Down Expand Up @@ -105,8 +105,8 @@ You can run the example with dbt debug log: `RUNTIME__LOG_LEVEL=DEBUG python dbt

## Other transforming tools

If you want to transform the data before loading, you can use Python. If you want to transform the data after loading, you can use dbt or one of the following:
If you want to transform your data before loading, you can use Python. If you want to transform your data after loading, you can use dbt or one of the following:

1. [`dlt` SQL client.](../sql.md)
2. [Pandas.](../pandas.md)
2. [Python with dataframes or arrow tables.](../python.md)

27 changes: 27 additions & 0 deletions docs/website/docs/dlt-ecosystem/transformations/index.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
---
title: Transforming your data
description: How to transform your data
keywords: [datasets, data, access, transformations]
---
import DocCardList from '@theme/DocCardList';

# Transforming data

If you'd like to transform your data after a pipeline load, you have 3 options available to you:

* [Using dbt](./dbt/dbt.md) - dlt provides a convenient dbt wrapper to make integration easier
* [Using the `dlt` SQL client](./sql.md) - dlt exposes an sql client to transform data on your destination directly using sql
* [Using python with dataframes or arrow tables](./python.md) - you can also transform your data using arrow tables and dataframes in python

If you need to preprocess some of your data before it is loaded, you can learn about strategies to:

* [Rename columns](../../general-usage/customising-pipelines/renaming_columns)
* [Pseudonymize columns](../../general-usage/customising-pipelines/pseudonymizing_columns)
* [Remove columns](../../general-usage/customising-pipelines/removing_columns)

This is particularly useful if you are trying to remove data related to PII or other sensitive data, you want to remove columns that are not needed for your use case or you are using a destination that does not support certain data types in your source data.


# Learn more
<DocCardList />

42 changes: 0 additions & 42 deletions docs/website/docs/dlt-ecosystem/transformations/pandas.md

This file was deleted.

109 changes: 109 additions & 0 deletions docs/website/docs/dlt-ecosystem/transformations/python.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
---
title: Transforming data in Python with arrow tables or dataframes
description: Transforming data loaded by a dlt pipeline with pandas dataframes or arrow tables
keywords: [transform, pandas]
---

# Transforming data in python with dataframes or arrow tables

You can transform your data in python using pandas dataframes or arrow tables. To get started, please read the [dataset docs](../../general-usage/dataset-access/dataset).


## Interactively transforming your data in python

Using the methods explained in the [dataset docs](../../general-usage/dataset-access/dataset), you can fetch data from your destination into a dataframe or arrow table in your local python process and work with it interactively. This even works for filesystem destinations:


The example below reads GitHub reactions data from the `issues` table and
counts the reaction types.

```py
pipeline = dlt.pipeline(
pipeline_name="github_pipeline",
destination="duckdb",
dataset_name="github_reactions",
dev_mode=True
)

# get a dataframe of all reactions from the dataset
reactions = pipeline.dataset().issues.select("reactions__+1", "reactions__-1", "reactions__laugh", "reactions__hooray", "reactions__rocket").df()

# calculate and print out the sum of all reactions
counts = reactions.sum(0).sort_values(0, ascending=False)
print(counts)

# alternatively, you can fetch the data as an arrow table
reactions = pipeline.dataset().issues.select("reactions__+1", "reactions__-1", "reactions__laugh", "reactions__hooray", "reactions__rocket").arrow()
# ... do transformations on the arrow table
```

## Persisting your transformed data

Since dlt supports dataframes and arrow tables from resources directly, you can use the same pipeline to load the transformed data back into the destination.


### A simple example

A simple example that creates a new table from an existing user table but only with columns that do not contain private information. Note that we use the iter_arrow() method on the relation to iterate over the arrow table instead of fetching it all at once.

```py
pipeline = dlt.pipeline(
pipeline_name="users_pipeline",
destination="duckdb",
dataset_name="users_raw",
dev_mode=True
)

# get user relation with only a few columns selected, but omitting email and name
users = pipeline.dataset().users.select("age", "amount_spent", "country")

# load the data into a new table called users_clean in the same dataset
pipeline.run(users.iter_arrow(chunk_size=1000), table_name="users_clean")
```

### A more complex example

The example above could easily be done in SQL. Let's assume you'd like to actually do some in python arrow transformations. For this will create a resources from which we can yield the modified arrow tables. The same is possibly with dataframes.

```py
import pyarrow.compute as pc

pipeline = dlt.pipeline(
pipeline_name="users_pipeline",
destination="duckdb",
dataset_name="users_raw",
dev_mode=True
)

# NOTE: this resource will work like a regular resource and support write_disposition, primary_key, etc.
# NOTE: For selecting only users above 18, we could also use the filter method on the relation with ibis expressions
@dlt.resource(table_name="users_clean")
def users_clean():
users = pipeline.dataset().users
for arrow_table in users.iter_arrow(chunk_size=1000):

# we want to filter out users under 18
age_filter = pc.greater_equal(arrow_table["age"], 18)
arrow_table = arrow_table.filter(age_filter)

# we want to hash the email column
arrow_table = arrow_table.append_column("email_hash", pc.sha256(arrow_table["email"]))

# we want to remove the email column and name column
arrow_table = arrow_table.drop(["email", "name"])

# yield the transformed arrow table
yield arrow_table


pipeline.run(users_clean())
```

## Other transforming tools

If you want to transform your data before loading, you can use Python. If you want to transform the
data after loading, you can use Pandas or one of the following:

1. [dbt.](dbt/dbt.md) (recommended)
2. [`dlt` SQL client.](sql.md)

Loading
Loading