dlt-hub / dlt-hub/dlt

Unexpected write_disposition behavior when moving data from S3 to Databricks

Open
#2,252 4 comments 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
5.9k
Forks
600
Avg merge
1d 14h
Merged PRs (30d)
38

Description

### dlt version

1.5.0

### Describe the problem

When write_disposition is set as a hint on the source like below it is ignored (defaults to append),

```
s3_orders = filesystem(
bucket_url="s3://oc-test-data-load-tool-dbx/oc_data_raw/orders",
file_glob="*.jsonl"
).apply_hints(incremental=dlt.sources.incremental("modification_date"), write_disposition="replace")
```

and when write_disposition is set at the pipeline level, it is used but incremental loading is ignored.
```
pipeline = dlt.pipeline(
import_schema_path="schemas/import",
export_schema_path="schemas/export",
pipeline_name="s3_to_databricks",
destination="databricks",
staging=dlt.destinations.filesystem("s3://oc-test-data-load-tool-dbx/oc_data_staging"),
dataset_name="oc_data_bronze",
dev_mode=False,
progress="tqdm"
)
orders_load_info = pipeline.run((s3_orders | read_jsonl()).with_name('orders'), write_disposition="replace")
logging.info(orders_load_info)
```

### Expected behavior

I expect the script to do a destructive replace when write_disposition="replace" is set at the source level, and I expect DLT to correctly use incremental loading even when the write_disposition is set at the pipeline level. (I do not expect behavior to be different based on whether write_disposition is set at the pipeline level or the source level)

### Steps to reproduce

Here is my full code with the write_disposition set as part of a hint: (Note that I am not running it as a lambda function, I am running on my local machine)

```import dlt
from dlt.sources.filesystem import filesystem, read_jsonl
import logging
from datetime import datetime

def lambda_handler(event={}, context=None):
s3_orders = filesystem(
bucket_url="s3://oc-test-data-load-tool-dbx/oc_data_raw/orders",
file_glob="*.jsonl"
).apply_hints(incremental=dlt.sources.incremental("modification_date"), write_disposition="append")

s3_order_line_items = filesystem(
bucket_url="s3://oc-test-data-load-tool-dbx/oc_data_raw/order_line_items",
file_glob="*.jsonl"
).apply_hints(incremental=dlt.sources.incremental("modification_date"), write_disposition="append")

s3_promotions = filesystem(
bucket_url="s3://oc-test-data-load-tool-dbx/oc_data_raw/promotions",
file_glob="*.jsonl"
).apply_hints(incremental=dlt.sources.incremental("modification_date"), write_disposition="replace")

s3_products = filesystem(
bucket_url="s3://oc-test-data-load-tool-dbx/oc_data_raw/products",
file_glob="*.jsonl"
).apply_hints(incremental=dlt.sources.incremental("modification_date"), write_disposition="replace")

s3_customers = filesystem(
bucket_url="s3://oc-test-data-load-tool-dbx/oc_data_raw/customers",
file_glob="*.jsonl"
).apply_hints(incremental=dlt.sources.incremental("modification_date"), write_disposition="replace")

pipeline = dlt.pipeline(
import_schema_path="schemas/import",
export_schema_path="schemas/export",
pipeline_name="s3_to_databricks",
destination="databricks",
staging=dlt.destinations.filesystem("s3://oc-test-data-load-tool-dbx/oc_data_staging"),
dataset_name="oc_data_bronze",
dev_mode=False,
progress="tqdm"
)

promotions_load_info = pipeline.run((s3_promotions | read_jsonl()).with_name('promotions'))
logging.info(promotions_load_info)
products_load_info = pipeline.run((s3_products | read_jsonl()).with_name('products'))
logging.info(products_load_info)
orders_load_info = pipeline.run((s3_orders | read_jsonl()).with_name('orders'))
logging.info(orders_load_info)
order_line_items_load_info = pipeline.run((s3_order_line_items | read_jsonl()).with_name('order_line_items'))
logging.info(order_line_items_load_info)
customers_load_info = pipeline.run((s3_customers | read_jsonl()).with_name('customers'))
logging.info(customers_load_info)

if __name__=="__main__":
lambda_handler()
```

The easiest way to replicate this would be to put a simple test json file into a bucket in s3, run the pipeline, then put an additional file into the s3 bucket, and run the pipeline again (write_disposition should be set to replace). You will see two _dlt_load_id's in the data:

```
select _dlt_load_id, count(*) FROM oc_data_bronze.promotions
GROUP BY _dlt_load_id;
```

To see the other error, you'll need to move the write_disposition to the pipeline level as shown below (I have a couple with write_disposition set to append but those can be ignored):

```
import dlt
from dlt.sources.filesystem import filesystem, read_jsonl
import logging
from datetime import datetime

def lambda_handler(event={}, context=None):
s3_orders = filesystem(
bucket_url="s3://oc-test-data-load-tool-dbx/oc_data_raw/orders",
file_glob="*.jsonl"
).apply_hints(incremental=dlt.sources.incremental("modification_date"))

s3_order_line_items = filesystem(
bucket_url="s3://oc-test-data-load-tool-dbx/oc_data_raw/order_line_items",
file_glob="*.jsonl"
).apply_hints(incremental=dlt.sources.incremental("modification_date"))

s3_promotions = filesystem(
bucket_url="s3://oc-test-data-load-tool-dbx/oc_data_raw/promotions",
file_glob="*.jsonl"
).apply_hints(incremental=dlt.sources.incremental("modification_date"))

s3_products = filesystem(
bucket_url="s3://oc-test-data-load-tool-dbx/oc_data_raw/products",
file_glob="*.jsonl"
).apply_hints(incremental=dlt.sources.incremental("modification_date"))

s3_customers = filesystem(
bucket_url="s3://oc-test-data-load-tool-dbx/oc_data_raw/customers",
file_glob="*.jsonl"
).apply_hints(incremental=dlt.sources.incremental("modification_date"))

pipeline = dlt.pipeline(
import_schema_path="schemas/import",
export_schema_path="schemas/export",
pipeline_name="s3_to_databricks",
destination="databricks",
staging=dlt.destinations.filesystem("s3://oc-test-data-load-tool-dbx/oc_data_staging"),
dataset_name="oc_data_bronze",
dev_mode=False,
progress="tqdm"
)

promotions_load_info = pipeline.run((s3_promotions | read_jsonl()).with_name('promotions'), write_disposition="replace")
logging.info(promotions_load_info)
products_load_info = pipeline.run((s3_products | read_jsonl()).with_name('products'), write_disposition="replace")
logging.info(products_load_info)
orders_load_info = pipeline.run((s3_orders | read_jsonl()).with_name('orders'), write_disposition="append")
logging.info(orders_load_info)
order_line_items_load_info = pipeline.run((s3_order_line_items | read_jsonl()).with_name('order_line_items'), write_disposition="append")
logging.info(order_line_items_load_info)
customers_load_info = pipeline.run((s3_customers | read_jsonl()).with_name('customers'), write_disposition="replace")
logging.info(customers_load_info)

if __name__=="__main__":
lambda_handler()
```

Run the pipeline with one file in the source S3 bucket, then add another and run again. You will see that the target table in Databricks has data from both S3 files, but only one single _dlt_load_id, implying that a destructive replace was performed, but both files were read when only the latest one should have been. You can also further confirm this in the logs for the second run by noting that start_value is set to None (it should be the date of your first load).

> {"written_at":"2025-01-30T17:01:21.635Z","written_ts":1738256481635656000,"component_name":"s3_to_databricks","process":12821,"msg":"Bind incremental on filesystem_products with initial_value: None, start_value: None, end_value: None","type":"log","logger":"dlt","thread":"MainThread","level":"INFO","module":"__init__","line_no":483,"version":{"dlt_version":"1.5.0","pipeline_name":"s3_to_databricks"}}

### Operating system

macOS

### Runtime environment

Local

### Python version

3.11

### dlt data source

S3 bucket

### dlt destination

_No response_

### Other deployment details

Destination is Databricks

### Additional information

_No response_

Contributor guide

Open the contributing guide

Research direction

Reproduce the two cases using the filesystem source, apply_hints(incremental=...), and pipeline.run(..., write_disposition=...) entry points described in the issue. Compare the second-run incremental state and target rows for source-level versus pipeline-level disposition. Done means replace removes prior target data while incremental loading still reads only new S3 files, regardless of where the disposition is set.

Written by the indexing model from the issue text.

Assessment

Tech stack
aws, python
Domain
data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.