data sink destination #891

sh-rp · 2024-01-12T16:20:19Z

Description

Implementation of #752

Notes:

We also introduce the concept of a load package state here

netlify · 2024-01-12T16:20:24Z

✅ Deploy Preview for dlt-hub-docs canceled.

Name	Link
🔨 Latest commit	`2db3430`
🔍 Latest deploy log	https://app.netlify.com/sites/dlt-hub-docs/deploys/65e99a28e46f940008217fbe

rudolfix · 2024-01-21T19:13:52Z

dlt/common/data_types/type_helpers.py

+            try:
+                return json.loads(value)
+            except Exception:
+                pass


we should raise ValueError I suppose? so the calling code knows that value cannot be coerced?

Maybe JSON DecodeError is better for debugging?

dlt/destinations/impl/sink/configuration.py

rudolfix · 2024-01-21T19:17:37Z

dlt/destinations/decorators.py

+
+def sink(loader_file_format: TLoaderFileFormat = None, batch_size: int = 10) -> Any:
+    def decorator(f: TSinkCallable) -> TDestinationReferenceArg:
+        return _sink(credentials=f, loader_file_format=loader_file_format, batch_size=batch_size)


IMO this is the best place to extract the destination name and clean it up the same way we extract source/resource name in the relevant decorator?

another thing we should solve is to pass credentials and additional parameters to the sing function. we can reuse dlt.transformer decorator does exactly what we want here (wraps callable in another callable where only additional parameters are visible and uses with_config). we can take a look at this together or do this is another PR after merging this

I'm not sure what you mean with cleaning up the name? Do you mean extract it from the callable name or something like that?

Ok, i added it so it will take a name param or use the callable name if possible

dlt/destinations/impl/sink/factory.py

rudolfix · 2024-01-21T19:44:37Z

dlt/destinations/impl/sink/sink.py

+
+
+# TODO: implement proper state storage somewhere, can this somehow go into the loadpackage?
+job_execution_storage: Dict[str, int] = {}


IMO best would be to use existing pipeline state to keep it. exactly like we handle the source state. you may extend pipeline.py in common (pipeline_state) method. there are a few difficulties though:

state should be cleared when a new package starts to load

we actually must persist state after each batch. right now state stays in memory and is written at the end of the context manager in the pipeline.

another problem with the state is that does not implement state sync interface. so if working directory is wiped, all the state is lost (source and destination). this will prevent incremental loading from working.

btw. we could do state sync as another plugin. if specified state and schemas would be stored and restored via plugin. we could use buckets or some KV databases like dynamo db or wrap any destination as such plugin (for destinations that support state sync)

does sink work with staging destination?

rudolfix · 2024-01-21T20:30:50Z

dlt/destinations/impl/sink/sink.py

+        self._schema = schema
+        self._job_execution_storage = job_execution_storage
+
+        # TODO: is this the correct way to tell dlt to retry this job in the next attempt?


read the docstrings for LoadJob. particular instance will not be retried. "retry" is a terminal state for the LoadJob class. loader.py will create a new instance of SinkLoadJob will be created

dlt/destinations/__init__.py

dlt/common/pipeline.py

rudolfix · 2024-01-25T19:55:24Z

dlt/common/pipeline.py

+
+
+def commit_pipeline_state() -> None:
+    container = Container()


make sure this code is thread safe! sinks are called from many threads and work in parallel

i did this in the new code in the place where the context is created, do you think we should create the lock here?

rudolfix · 2024-01-25T19:57:14Z

dlt/destinations/impl/sink/sink.py

            self._state = "completed"
        except Exception as e:
            self._state = "retry"
            raise e
+        finally:
+            # save progress
+            commit_pipeline_state()


hmmmm this looks good enough. better would be to save state after each batch.

rudolfix · 2024-01-25T20:03:28Z

dlt/destinations/impl/sink/sink.py

@@ -193,23 +144,21 @@ def update_stored_schema(
        return super().update_stored_schema(only_tables, expected_update)

    def start_file_load(self, table: TTableSchema, file_path: str, load_id: str) -> LoadJob:
+        load_state = destination_state().setdefault(load_id, {})


set default is not thread safe. but there won't be anyone else accessing this key so OK

sultaniman · 2024-01-30T12:49:15Z

dlt/common/configuration/providers/airflow.py

@@ -14,7 +14,7 @@ def name(self) -> str:

    def _look_vault(self, full_key: str, hint: type) -> str:
        """Get Airflow Variable with given `full_key`, return None if not found"""
-
+        return


Oops I for the moment thought that I committed this in my PR

dlt/common/storages/load_package.py

sh-rp · 2024-01-30T14:32:36Z

dlt/load/load.py

@@ -556,12 +566,29 @@ def run(self, pool: Optional[Executor]) -> TRunMetrics:
        schema = self.load_storage.normalized_packages.load_schema(load_id)
        logger.info(f"Loaded schema name {schema.name} and version {schema.stored_version}")

+        # prepare load package state context


i am injecting the load package state context here, i think it makes the most sense but it also clutters the load class a bit, wdyt? For now we only use the load package state for the destinations, so there is no need to implement this any other stage than load.

Now when I look at this I think we do not need to inject state at all, what we could inject is load id and instance of the package storage as a context available to a function decorated with sink. we can do that later.

sh-rp · 2024-01-30T14:37:04Z

@rudolfix you can have another look at this PR. The main questions are:

Is the load package state done right? Should we add a schema there and combine it with this file that for now only has one line of text (loaded etc) in it?
Should the sink function get a bit more info than just dataitems and table schema? Maybe sentinel values that indicate start and finish of a job or something like that? A context object could also be nice, but maybe no-one needs it.

sultaniman · 2024-01-30T15:10:29Z

dlt/common/storages/load_package.py

+            )
+            return cast(DictStrAny, json.loads(state))
+        except FileNotFoundError:
+            return {}


Is logging details is useful for debugging?

no, this just means that there is no statefile yet

sultaniman · 2024-02-07T09:37:59Z

dlt/common/data_types/type_helpers.py

+            try:
+                return json.loads(value)
+            except Exception:
+                raise ValueError("Cannot load text as json for complex type")


Should we make the message more clear because for me "Cannot" sounds like "we do not support", maybe "were unable to load text as json" or "Was unable to load text as json" or better?

the pattern is to raise ValueError(value). you'll get the error message from exception.context

dlt/destinations/impl/sink/factory.py

sultaniman · 2024-02-07T09:57:03Z

dlt/destinations/impl/sink/sink.py

+        file_path: str,
+        config: SinkClientConfiguration,
+        schema: Schema,
+        load_package_state: Dict[str, int],


Should this have TLoadPackageState alias instead?

no, but i renamed it to make it more clear what is is :)

sultaniman · 2024-02-07T10:15:25Z

docs/website/docs/dlt-ecosystem/destinations/sink.md

+
+```python
+@dlt.sink(batch_size=10, loader_file_format="jsonl", name="my_sink")
+def sink(items: TDataItems, table: TTableSchema) -> None:


Should we add a comment clarifying where users can import our internal types from?

you can see that a couple of lines above this example

tests/common/test_versioned_state.py

sultaniman · 2024-02-07T10:32:25Z

tests/cases.py

+    if expect_filtered_null_columns:
+        for key, expected in expected_rows.items():
+            if expected is None:
+                assert db_mapping.get(key, None) is None


maybe better to check that key is not in key not in db_mapping?

no, the key could be there and set to none and that would also be ok i would say.

sultaniman · 2024-02-07T10:48:40Z

dlt/common/storages/load_package.py

+
+    """A section of state that does not participate in change merging and version control"""
+    destinations: NotRequired[Dict[str, Dict[str, Any]]]
+    """private space for destinations to store state relevant only to the load package"""


What is this style of docstrings called?

i have no idea, i took this from the way it is done in other places @rudolfix is this some convention?

this is unofficial convention of adding docstrings to fields. I think there's no PEP for that

dlt/common/storages/load_package.py

dlt/destinations/decorators.py

dlt/destinations/impl/sink/sink.py

rudolfix

logic LGTM! I think we have basics covered.

Should we rename sink to something else? Alex is using "publisher", we could also use "dlt.destination" but maybe that would be too big for a name?
could we add dlt.current.destination_state() and dlt.current.load_state()` that could be used in the decorated function. It may make sense to keep some state or be able to inspect full schema etc.
I will probably try wrap a sink function so additional configuration (ie. credentials) can be passed to it. IMO we can merge into destination factory better than making credentials a callable

rudolfix · 2024-02-07T14:19:10Z

dlt/common/data_types/type_helpers.py

+            try:
+                return json.loads(value)
+            except Exception:
+                raise ValueError("Cannot load text as json for complex type")


the pattern is to raise ValueError(value). you'll get the error message from exception.context

rudolfix · 2024-02-07T14:20:54Z

dlt/common/pipeline.py

@@ -599,6 +602,28 @@ class StateInjectableContext(ContainerInjectableContext):
        def __init__(self, state: TPipelineState = None) -> None: ...


+@configspec
+class LoadPackageStateInjectableContext(ContainerInjectableContext):


cool but why here? it is a part of package storage. and it is a big 👍 that it is decoupled from pipeline (it will help me to handle bad data for example)

i moved it, but i am not 100% that it is in the right location now..

rudolfix · 2024-02-07T14:22:06Z

dlt/common/storages/load_package.py

+
+
+class TLoadPackageState(TVersionedState, total=False):
+    created: int


hmmm we use ISO timestamps everywhere

rudolfix · 2024-02-07T14:23:26Z

dlt/common/storages/load_package.py

+    """Timestamp when the loadpackage was created"""
+
+    """A section of state that does not participate in change merging and version control"""
+    destinations: NotRequired[Dict[str, Dict[str, Any]]]


a load package will be processed by just one destination. the processing is destructive (we move files around, create followup jobs etc.) so I assume we just need `destination_state which is a dict?

rudolfix · 2024-02-07T14:30:45Z

dlt/destinations/impl/sink/sink.py

+                if start_batch > 0:
+                    start_batch -= 1
+                    continue
+                batch = record_batch.to_pylist()


hmmmmm. IMO just yield arrow here. Alex for example is pushing panda frames to salesforce

dlt/pipeline/pipeline.py

rudolfix · 2024-02-07T14:41:36Z

dlt/destinations/impl/sink/sink.py

+
+    def complete_load(self, load_id: str) -> None:
+        # pop all state for this load on success
+        clear_loadpackage_destination_state()


IMO it is better if we do not delete it. You can use it ie. to inspect how many rows were processed.

alright, changed!

…sink_decorator # Conflicts: # dlt/pipeline/current.py

sh-rp · 2024-02-08T11:56:36Z

@rudolfix i implemented all the changes, since puae-jsonl files are also stored with the .jsonl extension, i had to fix a place in the loader which does not seem super nice. If we wanted to do this properly I think we would have puae-json (the current format but renamed) and puae-jsonl (new format with individually encoded lines) and save them with a different exension then jsonl, and it would also be nice to have an abstracted reader class (similar to the write) where you just stick a file path and it emits rows or batches of rows, so that this decoding code is not in the sink destination. But we could do a later ticket for that, not sure.

sh-rp · 2024-02-28T12:49:14Z

@rudolfix there were no more comments here since my last changes, did you forget to publish your review maybe? Since you mentioned yesterday that you'd like to have some kind of callbacks for start and end load, i also had a suggestion for special sentinels in the sink function. We also make the decorator work for classes that then need to implement a certain interface where we have class methods for start and end load etc.

# Conflicts: # dlt/common/storages/load_package.py # dlt/destinations/impl/athena/athena.py # dlt/destinations/impl/bigquery/bigquery.py # dlt/load/load.py

… sink

rudolfix

this is getting so good!
we'll have another round of comments. now I was focused on generating proper config and destination signature

regarding tests:
please implement a BigQuery insert api sink, use #1037 for a reference. you can add this in bigquery specific tests and in examples. what it should do

accept GcpCredentials and credentials + some default parameters
insert json in batches
bigquery should be able to infer the schema

rudolfix · 2024-03-05T19:34:32Z

dlt/common/storages/load_package.py

+
+    """A section of state that does not participate in change merging and version control"""
+    destinations: NotRequired[Dict[str, Dict[str, Any]]]
+    """private space for destinations to store state relevant only to the load package"""


this is unofficial convention of adding docstrings to fields. I think there's no PEP for that

rudolfix · 2024-03-05T19:36:38Z

dlt/common/storages/load_package.py


 # folders to manage load jobs in a single load package
 TJobState = Literal["new_jobs", "failed_jobs", "started_jobs", "completed_jobs"]
 WORKING_FOLDERS: Set[TJobState] = set(get_args(TJobState))
-TLoadPackageState = Literal["new", "extracted", "normalized", "loaded", "aborted"]
+TLoadPackageStatus = Literal["new", "extracted", "normalized", "loaded", "aborted"]


why it got changed? this is state, not status. state as in state machine through which package travels

rudolfix · 2024-03-05T19:41:40Z

dlt/destinations/decorators.py

+from dlt.common.utils import get_callable_name
+
+
+def sink(


we agreed to name this destination right @VioletM ?

yes, this is what the decorator is called when importing it from dlt (see the tests). I can also rename this function though.

rudolfix · 2024-03-05T19:53:54Z

dlt/destinations/decorators.py

+) -> Any:
+    def decorator(f: TSinkCallable) -> TDestinationReferenceArg:
+        nonlocal name
+        if name is None:


Or something was not pushed or the ticket was not clear. What I'd like to do there is to take SinkClientConfiguration and use is as a base to synthesize user config.

resource_sections = (known_sections.DESTINATION, name) # standalone resource will prefer existing section context when resolving config values # this lets the source to override those values and provide common section for all config values for resources present in that source # for autogenerated spec do not include defaults conf_f = with_config( incr_f, spec=spec, sections=resource_sections, include_defaults=spec is not None, base_=SinkClientConfiguration # you need to implement this ) # get spec for wrapped function SPEC = get_fun_spec(conf_f) # store the standalone resource information _DESITNATIONS[f.__qualname__] = SourceInfo(SPEC, f, func_module)

Now what you should have in SPEC:

your base

arguments from decorated function that have defaults or dlt.config/dlt.secret.value

if user uses credentials you'll have user credentials automatically
in essence you have destination configuration that works same as other configurations

rudolfix · 2024-03-05T20:00:32Z

dlt/destinations/impl/sink/configuration.py

+
+
+@configspec
+class SinkClientCredentials(CredentialsConfiguration):


Those are not credentials. User will provide credentials (eventually) in the decorated function. if you need to keep the reference to decorated function, use some other field name

Alright, moved the callable into the regular config

rudolfix · 2024-03-05T20:04:15Z

dlt/destinations/impl/sink/configuration.py

+            *,
+            credentials: Union[SinkClientCredentials, TSinkCallable, str] = None,
+            loader_file_format: TLoaderFileFormat = "puae-jsonl",
+            batch_size: int = 10,


file_format and batch_size are part of the decorator, not destination signature, remove

I'm not quite sure what you mean. When I create a destination in the decorator, I need a way to hand over these values to the destination instance, and in all destinations this kind of stuff is done via the config, no?

rudolfix · 2024-03-05T20:04:48Z

dlt/destinations/impl/sink/sink.py

+        try:
+            if self._config.batch_size == 0:
+                # on batch size zero we only call the callable with the filename
+                self.call_callable_with_items(self._file_path)


this is super good

rudolfix · 2024-03-05T20:06:30Z

dlt/destinations/impl/sink/sink.py

+        ) == 0, "Batch size was changed during processing of one load package"
+
+        start_batch = start_index / self._config.batch_size
+        with pyarrow.parquet.ParquetFile(self._file_path) as reader:


mhhhh this looks to me like file_readers - opposite to our file_writers. so you can read parquet and jsonl (with pua) / json with them.

makes sense to extract it now?

Let's do a separate PR for this where filetypes are fully pluggable..

rudolfix · 2024-03-05T20:09:24Z

dlt/destinations/impl/sink/sink.py

+                for batch in self.run(current_index):
+                    self.call_callable_with_items(batch)
+                    current_index += len(batch)
+                    destination_state[self._storage_id] = current_index


are you able to store batch number per table in state? (you'll need to lock the state). then user will be able to figure out that a batch is a first batch for a given table. you already store stuff in state

rudolfix · 2024-03-05T20:13:21Z

dlt/__init__.py

@@ -62,6 +64,7 @@
    "resource",
    "transformer",
    "defer",
+    "destination",


please also import load_package() into current module so it can be requested from within decorated function

It is there, no?

sh-rp · 2024-03-06T14:50:01Z

dlt/common/configuration/container.py

-                thread_id = int(m.group(1))
-            else:
-                thread_id = threading.get_ident()
+            # print(threading.currentThread().getName())


I'm not quite sure how this is supposed to work, in any case it will not work to inject multiple configcontexts of the same type in different threads if I keep this in. lmk.

sh-rp force-pushed the d#/data_sink_decorator branch 5 times, most recently from 9056025 to 9947a62 Compare January 15, 2024 16:24

rudolfix reviewed Jan 21, 2024

View reviewed changes

sh-rp linked an issue Jan 22, 2024 that may be closed by this pull request

implement destination sinks #752

Closed

sultaniman reviewed Jan 22, 2024

View reviewed changes

dlt/destinations/__init__.py Outdated Show resolved Hide resolved

rudolfix reviewed Jan 25, 2024

View reviewed changes

sh-rp added 9 commits January 30, 2024 12:36

start sink

8b6028b

parquet sink prototype

6a924d9

some more sink implementations

83c1af6

finish first batch of helpers

b972f23

add missing tests and fix linting

9dabeff

make configuratio more versatile

af6defd

implement sink function progress state

4d730e8

move to iterator

3b577b0

persist sink load state in pipeline state

5527689

sh-rp force-pushed the d#/data_sink_decorator branch from 8b3da5b to df5d025 Compare January 30, 2024 11:36

sultaniman reviewed Jan 30, 2024

View reviewed changes

fix unrelated typo

0657034

sh-rp force-pushed the d#/data_sink_decorator branch from df5d025 to 0657034 Compare January 30, 2024 12:52

move sink state storage to loadpackage state

b5db5b8

sh-rp commented Jan 30, 2024

View reviewed changes

dlt/common/storages/load_package.py Show resolved Hide resolved

sh-rp commented Jan 30, 2024

View reviewed changes

additional pr fixes

189d24b

disable creating empty state file on loadpackage init

57ed090

sultaniman reviewed Jan 30, 2024

View reviewed changes

fix linter error

376832d

sultaniman reviewed Feb 7, 2024

View reviewed changes

rudolfix requested changes Feb 7, 2024

View reviewed changes

sh-rp added 2 commits February 7, 2024 18:28

some pr fixes

79fce9e

Merge commit '17aea98527eab19f3300ee161114541f7eb2b5b5' into d#/data_…

8eb3e1a

…sink_decorator # Conflicts: # dlt/pipeline/current.py

more pr fixes

105569a

sh-rp force-pushed the d#/data_sink_decorator branch from 97b0eac to 105569a Compare February 8, 2024 12:11

small readme changes

27b8b2c

rudolfix mentioned this pull request Feb 29, 2024

add fast loading methods for bigquery destination #1037

Closed

sh-rp added 10 commits March 4, 2024 17:28

Merge branch 'devel' into d#/data_sink_decorator

e60f2f1

# Conflicts: # dlt/common/storages/load_package.py # dlt/destinations/impl/athena/athena.py # dlt/destinations/impl/bigquery/bigquery.py # dlt/load/load.py

add load id to loadpackage info in current

3229745

add support for directly passing through the naming convention to the…

dbbbe7c

… sink

add support for batch size zero (filepath passthrouh)

db9b488

use patched version of flak8 encoding

3c39f41

fix tests

3dfcf39

add support for secrets and config in sink

bc44618

update sink docs

db8d0ed

revert encodings branch

d8719c1

fix small linting problem

d7eb19d

sh-rp requested a review from rudolfix March 5, 2024 17:34

rudolfix requested changes Mar 5, 2024

View reviewed changes

add support for config specs

ef35502

sh-rp commented Mar 6, 2024

View reviewed changes

add possibility to create a resolved partial

2db3430

sh-rp merged commit f7ec07c into devel Mar 7, 2024
4 checks passed

This was referenced Mar 7, 2024

Revert "data sink destination" #1064

Closed

Generic destination / sink decorator #1065

Merged



		# TODO: implement proper state storage somewhere, can this somehow go into the loadpackage?
		job_execution_storage: Dict[str, int] = {}



		class TLoadPackageState(TVersionedState, total=False):
		created: int



		@configspec
		class SinkClientCredentials(CredentialsConfiguration):

data sink destination #891

data sink destination #891

Conversation

sh-rp commented Jan 12, 2024 • edited Loading

Description

netlify bot commented Jan 12, 2024 • edited Loading

✅ Deploy Preview for dlt-hub-docs canceled.

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sh-rp commented Jan 30, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rudolfix left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sh-rp commented Feb 8, 2024

sh-rp commented Feb 28, 2024

rudolfix left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sh-rp commented Jan 12, 2024 •

edited

Loading

netlify bot commented Jan 12, 2024 •

edited

Loading