Tap mongodb on the SDK, minimizing boilerplate is nice ⭐ Very little code and yet the feature-complete enough to pay big dividends https://github.com/z3z1ma/tap-mongodb Check it out!

Nice! We'll be (re)building tap-dynamodb in the next month or two and will likely emulate your approach here for that tap also.

Yes as long as it’s configurable because in the event I am pulling and writing data to blob storage, I like having the document as-is. But it is a reasonable option to impose at least a little structure. So yes, I agree. I’ll think it through.

Cool. Yeah, agreed. Definitely if exact schema is known or can be known via document scanning, then it's much cleaner to just send the exact schema of the documents/records.

Nice finally! @alexander_butler, I think I still have a PR request hanging for the other one.

Am I understanding correctly that this tap is using collection.find, and not using change streams? https://github.com/z3z1ma/tap-mongodb/blob/main/tap_mongodb/collection.py#LL88

Yep you got it, as your link shows. There are a number of reasons I could go into as to why we aren't using change streams right now, though I explored it. I'm not averse to it but as a capped collection, the oplog will purge itself in order to not grow unboundedly. It makes it more complex (and in many cases a hard stop) for a batch load use case. We use mongo pretty extensively at harness.io and it would not have been immediately useful. It is much better for batch loading data to rely on a timestamp field in the document which any large collection of mutable documents ought contain.

Thanks. We use MongoDB and AWS DocumentDB - the latter does not support an oplog, only change streams. Our current analytics database (which we are actively deprecating) is actually fed by the mongo oplog - it’s been problematic for us as oplog documents can take different forms depending on whether an update in mongo used the “update document” syntax or the “replacement document” syntax. (So if an application changes from

num: num + 1

to

{ $inc: num }

, a new column is added in our analytics db and we have to update the logic there. If we were using change streams, there’d be no change. Not that this is a criticism of your work! It’s great - it just maybe won’t support my use case.

Makes perfect sense. As far as I understand it, change streams are akin to an abstraction over oplogs and when you no longer have data in the oplog beyond a certain date, you are also not able to use change streams before that date from my working with them. So to clarify I am using neither change streams nor oplogs. But for your use case you would ideally want _____?

As far as I understand it, change streams are akin to an abstraction over oplogs and when you no longer have data in the oplog beyond a certain date, you are also not able to use change streams before that date from my working with them.

That’s correct to my understanding as well. More context on my situation: we currently have a Java application that subscribes to mongodb change streams and publishes that data to a Kafka topic, and then it’s written to S3 by a Confluent S3 Sink connector. Then from S3 we’re starting to use meltano to pull it into AWS Redshift. I’m wondering if it’s feasible to have meltano pull data from the change stream and push to redshift - in theory that’d let us remove the Java application, the kafka topic, and s3 in one fell swoop.

you are also not able to use change streams before that date from my working with them.

This is my understanding as well, although I’ve been unable to find clear/conclusive documentation that says this. This is something that our current Java service provides an abstraction over: the change streams are configured to return the full document, not just the changed fields. And the java application supports restarting the changestream from arbitrary points (to do a full initial load, for instance) by just returning documents from the collection based on the timestamp in the

_id

ObjectId field.

But for your use case you would ideally want _____?

ideally for me, I think it’d be “provide change stream data (with the change stream configured to return the full document), unless you are requesting data from a time period where there is no change stream data available, then provide a stream of documents from `.find`”

And the java application supports restarting the changestream from arbitrary points (to do a full initial load, for instance) by just returning documents from the collection based on the timestamp in the

_id

ObjectId field.

This is basically what I imagined we could do in the tap too. I just didn't implement it yet because the complexity vs the benefits for us personally wasn't there. But I am happy to take and collaborate on a PR if you're invested enough because the tap isn't far off from being able to do the same thing. Otherwise you can file an issue and I'll backlog it and tackle it when I have time 🙂 There are some edge cases to think through I think but otherwise it may not be too bad.

I will be happy to try out the tap and noodle on it a bit - thanks!

Yes I can totally see this being useful for us too at some point. If we can log an issue on the repo with the meat of the above 2 points you made:

RE: This is something that our current Java service provides an abstraction over: the change streams are configured...

RE: ideally for me, I think it’d be “provide change stream data (with the change stream configured to return the full document), unless you are requesting data from a time period where there is no change stream data available, then provide a stream of documents from

.find

I think I can take the lead on a PR and open it up to collaboration after hacking a bit. Also totally open if you fork the repo and take the lead. Either or, will collab in either case.

Wow. Nice work! I'll definitely take a look.

Hi @aaronsteers, out of curiosity - are there still plans for the Meltano team to (re)build a new version of

tap-dynamodb

in the works? We’re considering adding that to our implementation, but most of the existing variants appear to have some quirks and don’t look to be particularly actively maintained. If Meltano itself is considering rolling out a new version in the near future that might fill some of those gaps, that’d be super helpful to know - thanks!

Hi, @nick_hamlin. Definitely! We still are, and yes, we'd love to work together with you on it.

Hi @aaronsteers, thanks for that confirmation - I’ve been doing some initial testing with the singer variant and I think that’ll work fine for the short term, but naturally we’d like to avoid adding dependencies to the stack that don’t have longer-term maintenance plans

Sounds good! I used the Singer variant for a bit. The one caveat I can share upfront (at least from my usage a while back) is that you'll need to add a custom catalog with at least the top-level properties defined. The tap will otherwise just give a top-level {"type": "object"} definition per stream, which is not enough for targets to create the destination table.

Oh interesting, that’s a good callout - thanks for the heads up!

Do you mean in

meltano.yml

? Yes, if so. That basically sends your provided info into the catalog as overrides.

yep, cool, that makes sense (technically we have our taps segmented into their own separate files to keep

meltano.yml

from getting unwieldy, but yes)

If I could also put a vote in for adding support for

INCREMENTAL

replication, that would also be awesome 🙂

But back to your immediate use case, if you are fine to manually provide the schema while using the Singer variant, I otherwise had a good experience with the Singer variant. I can also recommend to use

LOG_BASED

replication from Singer's implementation. That option uses DynamoDB Streams to implement CDC without data loss if your incremental column is unreliable (which most inevitably are).

Excellent, thanks for the guidance! I’ll report back as we continue to explore

Awesome! I'm mindful of the fact that we've got a lot of folks on this thread, and we'll certainly want to revisit some of these topics. Here's a new channel for just that: • #C04TSH483DF

Hey @nick_hamlin, can you please share how you’re storing the schema files?

@abhishek_ajmera, in the experimenting I’ve done previously with other tap-dynamo incarnations, I’ve stored the schema in yaml alongside the other config. Does that answer your question? Happy to share more if helpful

So I’ve been storing the schema in catalog.json file, and I’m not very sure if I’m defining the schema in the best way or not. If you could share any example of schema of an attribute of type record/JSON/dictionary it would be great!

Sure, here’s an example of what I’ve got in

meltano.yml

for this:

plugins:
  extractors:
  - name: tap-dynamodb
    variant: singer-io
    capabilities:
      - catalog
      - discover
      - state
    config:
      account_id: <REDACTED>
      external_id: <REDACTED>
      role_name: <REDACTED>
      region_name: <REDACTED>
    metadata:
      my_dynamo_table:
        replication-method: FULL_TABLE
    schema:
      my_dynamo_table:
        PK:
          type: string
        SK:
          type: string
        field_1:
          type: number
        field_2:
          type: string 
        field_3:
          type: number
    select:
    - my_dynamo_table.*

Thankyou so much!