diff --git a/docs/changelog.md b/docs/changelog.md index 68c0175d5..fd4d92911 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -3,6 +3,7 @@ ## in progress - Connectors: Updated to Asana client v5 +- Connectors: Added source and target adapters for the Delta Lake table format ## 2026/06/25 v0.4.0 diff --git a/omniload/core/registry.py b/omniload/core/registry.py index 6392093a9..e97fafafc 100644 --- a/omniload/core/registry.py +++ b/omniload/core/registry.py @@ -42,6 +42,7 @@ "appstore": "omniload.source.appstore.api:AppleAppStoreSource", "asana": "omniload.source.asana.api:AsanaSource", "attio": "omniload.source.attio.api:AttioSource", + "az+delta": "omniload.source.deltalake.api:DeltaLakeSource", "bruin": "omniload.source.bruin.api:BruinSource", "chess": "omniload.source.chess.api:ChessSource", "clickup": "omniload.source.clickup.api:ClickupSource", @@ -54,6 +55,7 @@ "dynamodb": "omniload.source.dynamodb.api:DynamoDBSource", "elasticsearch": "omniload.source.elasticsearch.api:ElasticsearchSource", "facebookads": "omniload.source.facebook_ads.api:FacebookAdsSource", + "file+delta": "omniload.source.deltalake.api:DeltaLakeSource", "fireflies": "omniload.source.fireflies.api:FirefliesSource", "fluxx": "omniload.source.fluxx.api:FluxxSource", "frankfurter": "omniload.source.frankfurter.api:FrankfurterSource", @@ -64,7 +66,9 @@ "googleanalytics": "omniload.source.google_analytics.api:GoogleAnalyticsSource", "gorgias": "omniload.source.gorgias.api:GorgiasSource", "gs": "omniload.source.blobstorage.api:GCSSource", + "gs+delta": "omniload.source.deltalake.api:DeltaLakeSource", "gsheets": "omniload.source.google_sheets.api:GoogleSheetsSource", + "hdfs+delta": "omniload.source.deltalake.api:DeltaLakeSource", "hostaway": "omniload.source.hostaway.api:HostawaySource", "http": "omniload.source.http.api:HttpSource", "https": "omniload.source.http.api:HttpSource", @@ -77,6 +81,7 @@ "kafka": "omniload.source.kafka.api:KafkaSource", "kinesis": "omniload.source.kinesis.api:KinesisSource", "klaviyo": "omniload.source.klaviyo.api:KlaviyoSource", + "lakefs+delta": "omniload.source.deltalake.api:DeltaLakeSource", "linear": "omniload.source.linear.api:LinearSource", "linkedinads": "omniload.source.linkedin_ads.api:LinkedInAdsSource", "mailchimp": "omniload.source.mailchimp.api:MailchimpSource", @@ -96,6 +101,7 @@ "redditads": "omniload.source.reddit_ads.api:RedditAdsSource", "revenuecat": "omniload.source.revenuecat.api:RevenueCatSource", "s3": "omniload.source.blobstorage.api:S3Source", + "s3+delta": "omniload.source.deltalake.api:DeltaLakeSource", "salesforce": "omniload.source.salesforce.api:SalesforceSource", "sftp": "omniload.source.sftp.api:SFTPSource", "shopify": "omniload.source.shopify.api:ShopifySource", @@ -107,6 +113,7 @@ "stripe": "omniload.source.stripe.api:StripeAnalyticsSource", "tiktok": "omniload.source.tiktok_ads.api:TikTokSource", "trustpilot": "omniload.source.trustpilot.api:TrustpilotSource", + "uc+delta": "omniload.source.deltalake.api:DeltaLakeSource", "wise": "omniload.source.wise.api:WiseSource", "zendesk": "omniload.source.zendesk.api:ZendeskSource", "zoom": "omniload.source.zoom.api:ZoomSource", @@ -115,6 +122,7 @@ destinations: LazyRegistry = LazyRegistry( { + "az+delta": "omniload.target.deltalake:DeltaLakeDestination", "athena": "omniload.target.athena:AthenaDestination", "bigquery": "omniload.target.bigquery:BigQueryDestination", "clickhouse": "omniload.target.clickhouse:ClickhouseDestination", @@ -124,7 +132,11 @@ "databricks": "omniload.target.databricks:DatabricksDestination", "duckdb": "omniload.target.duckdb:DuckDBDestination", "elasticsearch": "omniload.target.elasticsearch.api:ElasticsearchDestination", + "file+delta": "omniload.target.deltalake:DeltaLakeDestination", "gs": "omniload.target.blobstorage:GCSDestination", + "gs+delta": "omniload.target.deltalake:DeltaLakeDestination", + "hdfs+delta": "omniload.target.deltalake:DeltaLakeDestination", + "lakefs+delta": "omniload.target.deltalake:DeltaLakeDestination", "md": "omniload.target.motherduck:MotherduckDestination", "mongodb": "omniload.target.mongodb:MongoDBDestination", "mongodb+srv": "omniload.target.mongodb:MongoDBDestination", @@ -139,6 +151,7 @@ "redshift+psycopg2": "omniload.target.redshift:RedshiftDestination", "redshift+redshift_connector": "omniload.target.redshift:RedshiftDestination", "s3": "omniload.target.blobstorage:S3Destination", + "s3+delta": "omniload.target.deltalake:DeltaLakeDestination", "snowflake": "omniload.target.snowflake:SnowflakeDestination", "sqlite": "omniload.target.sqlite:SqliteDestination", "synapse": "omniload.target.synapse:SynapseDestination", diff --git a/omniload/source/deltalake/adapter.py b/omniload/source/deltalake/adapter.py new file mode 100644 index 000000000..b1eeb08bb --- /dev/null +++ b/omniload/source/deltalake/adapter.py @@ -0,0 +1,53 @@ +from typing import Iterable, Optional + +import dlt +import polars as pl +from dlt.extract import DltResource +from yarl import URL + + +@dlt.source(name="deltalake", max_table_nesting=0) +def deltalake_source( + uri: str, + table: str, + batch_size: Optional[int] = 75_000, +) -> Iterable[DltResource]: + """ + Read from Delta Lake tables. + + Args: + uri (str): A filesystem URI that addresses the Delta Lake catalog. + table (str): . that addresses the Delta Lake table. + batch_size (int): Batch size for Polars + + Returns: + Iterable[DltResource]: Resources with data in RANDOM ORDER (optimized for speed). + """ + + url = URL(uri) + storage_options = dict(url.query) + url = url.with_query(None) + + if url.scheme != "uc": + table_fields = table.split(".") + if len(table_fields) != 2: + raise ValueError("Table name must be in the format .
") + url = url.joinpath(table_fields[-2], table_fields[-1]) + + uri = str(url) + + with pl.Config(streaming_chunk_size=batch_size): + + def reader(): + frame = pl.scan_delta(uri, storage_options=storage_options) + for i, batch in enumerate( + frame.collect_batches(engine="streaming", chunk_size=batch_size) + ): + yield batch + + return dlt.resource( + reader, + name=table, + # TODO: Are other write dispositions possible? + write_disposition="replace", + )() diff --git a/omniload/source/deltalake/api.py b/omniload/source/deltalake/api.py new file mode 100644 index 000000000..451a7b3c3 --- /dev/null +++ b/omniload/source/deltalake/api.py @@ -0,0 +1,17 @@ +class DeltaLakeSource: + def handles_incrementality(self) -> bool: + return True + + def dlt_source(self, uri: str, table: str, **kwargs): + + uri = uri.replace("+delta://", "://") + + # TODO: Review! + if kwargs.get("incremental_key"): + raise ValueError( + "DeltaLake takes care of incrementality on its own, you should not provide incremental_key" + ) + + from omniload.source.deltalake.adapter import deltalake_source + + return deltalake_source(uri, table) diff --git a/omniload/target/deltalake/__init__.py b/omniload/target/deltalake/__init__.py new file mode 100644 index 000000000..38f2ebe47 --- /dev/null +++ b/omniload/target/deltalake/__init__.py @@ -0,0 +1,20 @@ +import dlt +from yarl import URL + +from omniload.target.model import GenericSqlDestination + + +class DeltaLakeDestination(GenericSqlDestination): + def dlt_dest(self, uri: str, **kwargs): + kwargs.pop("dest_table", None) + kwargs.pop("staging_bucket", None) + uri = uri.replace("+delta://", "://") + # TODO: Review: Why not prune query parameters from URL when conveyed via `credentials`? + url = URL(uri) + creds = url.query + return dlt.destinations.filesystem(bucket_url=uri, credentials=creds, **kwargs) + + def dlt_run_params(self, uri: str, table: str, **kwargs) -> dict: + params = super().dlt_run_params(uri, table, **kwargs) + params["table_format"] = "delta" + return params diff --git a/pyproject.toml b/pyproject.toml index 967b9a59f..804dee025 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -82,7 +82,7 @@ dependencies = [ "databricks-sql-connector<5", "databricks-sqlalchemy<3", "dataclasses-json<0.7", - "dlt<1.29", + "dlt[az,deltalake,polars]<1.29", "dlt-cratedb<0.2", "duckdb<1.6", "duckdb-engine<0.18", @@ -97,6 +97,7 @@ dependencies = [ "google-cloud-bigquery-storage<3", "google-cloud-spanner<4", "influxdb-client<2", + "lakefs-spec", "mysql-connector-python<10", "oracledb<4", "orjson<4", @@ -131,6 +132,7 @@ dependencies = [ "trino[sqlalchemy]==0.337.0", "typer<0.27", "urllib3<2.8", + "yarl<2", ] optional-dependencies.develop = [ "pkginfo<2",