Over the holiday break I worked on tap-singer-jsonl and target-singer-jsonl, you know... to relax 😅 With my wife out in London tonight, I also had a chance to write up some thoughts on where I'd like to get them to 🚀 Somewhat of a throwback to my Spark/data lake days 🌊🚣‍♂️ Would love to hear any thoughts or ideas for improvements, especially with #C03569WJXBL coming up 🙌

Awesome I was literally drafting a plan that was almost identical! Very cool to see this. I also agree a tap and target are the right units of logic for this. I also agree on the data encoded into the layout of files in the bucket as being intuitive and right. The only difference was I imagined the target outputting/updating a metadata file that the corresponding tap uses instead of the tap scraping the bucket. Bucket operations are notoriously slow at scale and it is dependent on how many files the data lake is split into. So a central metadata file per tap can solve for that. Let me riff a bit so it makes sense. So

tap-salesforce | target-singer-jsonl

would produce a

tap-salesforce.singerlake

or some such metadata file. The subsequent

tap-singer-jsonl

only needs to point to this file to replicate a specific taps output instead of pointing to a path like

raw/<tap_id>/<stream_name>/*.jsonl

. A few reasons for this are listed below: Generally speaking the problem with parallelization would be that you don't know when the schema changes. We can only parallelize messages sharing the same schema safely since there would definitely be some sort of contention parallelizing conflicting schemas. This metadata files solves for that. The below illustrates the general idea.

{
  "schema_hashes": [...], // chronologically sorted array of md5(schema_message) hashes, ie. append only updates
  "files": {
    "<hash>": [...] // files that can be massively parallelized  
    "<hash>": [...]
   }
}

A state file then looks like this which incidentally solves for not needing the scrape bucket metadata or recursively list files in order to parse timestamp ranges whilst keeping the state message tame.

{
  "bookmarks": {
    "current-hash": "<hash>", // the last processed md5(schema) hash
    "last-index": 123  // the last index of the associated array representing a file
  }
}

You can with a single file read + the state message, immediately start reading in parallel with a known queue size.

@ken_payne has a blog I didn't know about. Awesome stuff!

@alexander_butler thanks for checking it out, and your improvements makes a ton of sense 🙌 I have actually relied on S3 Inventory before to generate that list of files 🤔 Which reminds me of two other techniques we applied at the time: • Another option for tracking the schema changes is to encode it into the prefix:

raw/<tap_id>/<stream_name>/<schema version>/...

. I imagine this schema version would be relative rather than absolute (i.e. the first schema observed is

v1

and the first change is

v2

etc.). This is also common in API design. • For even more parallelism, we could encode the year/month/day into the prefix too. This essentially pre-chunks the data, as we already know the dimensions will be 'observed date' based:

raw/<tap_id>/<stream_name>/<schema version>/<YYYY>/<MM>/<DD>/<HH>/...

. I'd say the

<HH>

extent of chunking could be configurable - not every use case will necessitate

<HH>

or even

<DD>

granularity 😅 It means slightly more work for the

target

(as it has to start new files on the chunk boundaries) but saves the

tap

having to deal with files that contain records in the middle of its preferred chunk Both of these combined would reduce the metadata/inventory file to a flat list of file paths, from which parallelism decisions can be made 🙂 WDYT?

@ken_payne that’s money! Great ideas 👀. Only consideration is when we start sync number 2 onwards, we need to verify our schema message is indeed conformant with the v(n) api we are on. So there is still a need to store a hash map of schema hash to version I believe? But otherwise I think everything you described is great, I love the api version encoded into the path. It’s a succinct and elegant way to add a missing piece of context 🙌.

{
  "versions": {"<hash>": "v1", "<hash>": "v2", ...},
  "files": [...]
}

Its great timing that you are thinking about this too 🙌 I came back 'round to hashing the schema over lunch. My initial thought was to store the last file read (index or filename) and the schema hash as bookmark, but it strikes me as brittle; strange things would happen if the state backend and bookmark were lost or unavailable. A stored hash is much safer in that regard 🤔 Also, in an ideal world, we would want to be able to recreate the metadata file from the list of files as efficiently as possible, on the assumption that with many taps being written to the same singerlake, something is bound to fail before updating the metadata file. I.e. the metadata is an accelerator, but the spec can fall back onto a "full scan" or list in its absence. So I looked for a hash concise enough to be included in the path 😅 Googles' FarmHash has a fingerprint function that will reduce a string to an unsigned int64, which when base85 encoded yields 7 characters. This is more than short enough to be stored in the path/prefix 🤯

something is bound to fail before updating the metadata file. I.e. the metadata is an accelerator, but the spec can fall back onto a "full scan" or list in its absence.

Agree with this 💯. Metadata is an accelerator rather than a point-of-failure. ☝️ One difference though is I was thinking one singerlake file to one tap. This means one

tap-salesforce.singerlake

, one

tap-github.singerlake

, and so on. This reduces the chance of multiple taps writing to the same file. An implication of this which I did not mention is that the file path + a stem could double as a lock. We only need to add a

tap-salesforce.singerlake.lock

to the bucket to ensure another

target-jsonl-singer

pointing at the exact same metadata file will not try to sync concurrently. I think a locking mechanism is important. EDIT: Also by virtue of one tap to one metadata file, tap-jsonl-singer only needs to point to a specific file to replicate an instance of a tap vs a consolidated metadata file of all taps which may contend for updates. This means, ideally, updates are fairly safe in 99% of situations. That being said, the "full scan" when needed and the ability to reconstruct the metadata file if needed are both very important pieces. nod Lastly on this point,

raw/tap-salesforce.singerlake

raw/tap-salesforce.singerlake.lock

raw/tap-salesforce/account/<FarmHash_A>/...

raw/tap-salesforce/account/<FarmHash_B>/...

raw/tap-salesforce/contact/<FarmHash_B>/...

The lake might look like this. The

raw

(base path) should be configurable and based on environment ideally. This means prod / dev route to different places and to different metadata files / different locks. 🔒

Googles' FarmHash has a fingerprint function that will reduce a string to an unsigned int64, which when base85 encoded yields 7 characters.

Wow thats awesome. The implications are you dont need the hashmap in a metadata file because the correct write path is known. So the file is again just a list of files which is nice. 🔥

I wonder if a new

singerlake

package is a best next step?

I think this is a good idea. With that, in theory even without a tap or target could just have a BigQuery/Snowflake/or whatever loader dependent on

python-singerlake

that uses

.get_messages()

directly, manages its own state, and do away with the "single file descriptor everything must pass through (stdout)". thinkspin

Sounds great! My preference would be to start with python, for myself and as most of the Meltano community lean in that direction too 🐍