Saving state for Incremental Sync for File based Destinations

[ShahBinoy]

I had few questions about incremental change-set and managing incremental data from DBs to FileSystem (s3)

1. With Destination being parquet, how does Sling find the continuation mark to fetch the delta?
2. If the destination is in S3, and path is governed via the naming convention, is Sling capable of filling the data gap limited to only what's missing?
3. Does it need to store state somewhere to continue ?
4. If deploying on Stateless environments ( Kubernetes / Serverless ), is it possible to preserve the state file ( if any ) to S3, etc and resume operations from there ?

[flarco]

Hey, at the moment sling doesn't support this. if you run incremental mode with a file destination, it's not really incremental (just full-refresh). For that we'd need a watermark (or state somewhere) as you've suggested. Would need to think about how that would work. Thinking the target filesystem itself would be the place to put the state, like a json file keeping the latest update_key value for all the streams:

{
  "my_schema.my_table": "2023-01-24 12:34 56.6789",
  "my_schema.my_table2": "2023-01-24 12:34 56.6789",
  "my_custom_sql": "2023-01-24 12:34 56.6789"
}

Sling would read from it at the start of run, and overwrite it at end of run.
Perhaps as a target.options, incremental_state: s3://my-bucket/path/to/state.json.
Thoughts?

[flarco]

That could work with DB to DB, and file to DB as well 🤔 , but then would need to put connection info in the incremental_state entry, like "incremental_state" : { "conn": "MY_S3", "url": "s3://my-bucket/path/to/state.json" }

[ShahBinoy]

Saving to S3, etc shared locations could work.

Here again something that comes to my mind is Apache Iceberg where the Iceberg metadata format could also act as a format of saving incremental dataset

[ShahBinoy]

Hi @flarco, is there any thought on taking this further ?

Also i wanted to brainstorm on how I can perform incremental fetch of data from DB to Parquet. This is the use case:

• First day run, select * from table where date < today() # Get all data to parquet
• Second and further day run. select * from table where date between (today() and (today() -1)) # Use SQL date Arithmatic of some kind
  OR
• Each Day run assisted by ENV Variable select * from my custom_query_table where date between ${START_DATE} and ${END_DATE} # I can inject START_DATE and END_DATE during sling run time as environment variables

Would either of approach be possible if State management is a larger effort ?

[ShahBinoy]

That could work with DB to DB, and file to DB as well 🤔 , but then would need to put connection info in the incremental_state entry, like "incremental_state" : { "conn": "MY_S3", "url": "s3://my-bucket/path/to/state.json" }

Additionally, incremental state cannot be only for source, it should be unique to each pair of source and destination

[nixent]

Wouldn't saving state for file destination end up in implementation of basic functionality of iceberg or delta format?
Things to consider:

• Append only inserts
• Updates
• Deletes
• Concurrent writes
• Handling of discrepancy between state and data files

Basically ACID

[ShahBinoy]

I dont think I am talking about that state, the state I am looking for is for evaluation of delta for data fetch. ie I have sync'ed exported data till yesterday, I need to pull from today onwards.

[hmac]

It would be really great to support incremental loading for file targets.

My use case is loading data from mongodb to a parquet file in blob storage. For parquet specifically, I believe it stores min/max values for each column in the file, so it should even be efficient to query the file for, e.g. MAX(update_key) on startup.

But having a separate state file alongside would also be totally fine for my use case.

[hubmos]

Hello!

@flarco do you have any new insights to share regarding this feature? Is there any plans on covering the use case of incremental load to file (in particular parquet)?

Has anyone found a workaround, or is the only option to go DB to DB if one need incremental load?

[flarco]

State based incremental is available in latest version. See here.