Hey everyone! I am trying to write a new tap using Meltano features I haven't used before and the result is making it clear that I do not really understand how to use them. The tap brings in github copilot usage metrics (schema: https://docs.github.com/en/copilot/reference/copilot-usage-metrics/example-schema). As you can see in the schema, many of the fields are arrays (ex: totals_by_ide). For each of these arrays, I wanted to split them out into their own stream so each array gets its own table in the target. My current implementation is a parent stream gets the full records and passes the records in-full to the child streams:

def get_child_context(self, record: dict, context: Context | None) -> dict:
        """Return context for child streams."""
        child_context = {
            "org": record.get("org"),
            "day": record.get("day") or record.get("date"),
            "copilot_metrics_record": record,
        }
        if context:
            return {**context, **child_context}
        return child_context

The consequence is three fold: (1) The logging prints the context so it gets absolutely littered, (2) the parent stream records get sent over to the target (we only want the child stream records), and (3) the child streams begin anew for each record: > Beginning sync of 'copilot_usage_totals_by_ide' in full_table mode with context: ... As you can see, clearly my understanding of how this works is lacking but I am not sure what other options are available to meet this goal. Does anyone have any advice or example projects that do something similar? EDIT: I found documentation on inline stream maps that could offer a different path: https://sdk.meltano.com/en/latest/stream_maps.html#duplicating-or-splitting-a-stream-using-source My only issue is I would prefer to define everything in the tap itself rather than

meltano.yml

so it is consistent with our other taps.

For 1, previously I have implemented a

hiddendict

to wrap a record passed via context and obfuscate its contents in the logs as `***`: • https://github.com/Matatika/tap-aptem/blob/712fc0870f79a759b2afddaff2dfce0a93622d72/tap_aptem/__init__.py#L8 • https://github.com/Matatika/tap-aptem/blob/712fc0870f79a759b2afddaff2dfce0a93622d72/tap_aptem/client.py#L143 You'll also want to set

stream_partitioning_keys = ()

on your child stream classes to prevent an explosion of bookmark contexts in state, as a result of having been passed an entire record. 2 I'm not sure what you mean? Only data synced by a stream (as per defined schemas) will be consumed by a target. 3 I think you can solve if you can make the parent stream incremental - child streams will then implicitly behave incrementally also. If the parent stream is already incremental, I don't think this is an issue (although logs may indicate otherwise, i.e.

in full_table mode

).

Hey @Reuben (Matatika), I appreciate the response. I will look at your

hiddendict

implementation and see if that will provide a solution! I have never made use of the built-in state capabilities of Meltano but if I ever do, I will look into

stream_partitioning_keys=()

as well. In regards to (2), it is probably confusing because it is a design flaw on my part that is difficult for me to express due to lack of understanding. My design at the time of writing my comment was the "parent" stream retrieved the data and then passed the full records to the child streams. Each child stream would take the array it was designed for from the record and sync it to the target. However, I do not want that high level record that the "parent" stream handled to be loaded into the target. So (2) was asking how I can prevent the "parent" stream from syncing its records over but still have the children. I do not expect a solution to this because that goes fundamentally against Meltano's design and has since been scrapped. I was essentially trying to use the "parent" as an intermediate data source that the children read from. For (3), it was set to incremental but I discovered that the parent-child model is really designed as a way to send a small list of records (containing parameters) and the child syncs once per record. There is no way to make the child collect all of the records first and then process them because that is not its intended design (at least that is my understanding). If the parent sends over 10k records to the child, that means the child stream will start up and shut down 10k times and same with the target. That is a lot of waste.

However, I do not want that high level record that the "parent" stream handled to be loaded into the target.

You could set

selected = False

or

selected_by_default = False

on the parent stream class: • selected • selected_by_default

There is no way to make the child collect all of the records first and then process them because that is not its intended design (at least that is my understanding). If the parent sends over 10k records to the child, that means the child stream will start up and shut down 10k times and same with the target. That is a lot of waste.

Correct 👍 I don't know if I would call it wasteful, but certainly there are cases when you want to collect child contexts together and apply them all at once (and have implemented this as a buffer system in a couple of taps - definitely hacky).@Edgar Ramírez (Arch.dev) may have more insight here.

The reason I called it waste is because our target is Redshift which handles small and rapid inserts very poorly. 10k+ instances of a Redshift DB connection being made, inserting 3 records, closing, and repeat is where the real waste is. I am actually not using

requests-cache

for my solution (never heard of it!). I am being a brute and just making the call and storing the parsed result in a class scope variable. Let me look into that! That may help bring back the elegance I was hoping for. Greatly appreciate the tip!

At one point I shipped feat(taps): Queue parent contexts and sync child streams only when the queue is full, but reverted as it was causing issues with the order of Singer messages some taps (specially MeltanoLabs/tap-github) expect. I'd like to revisit it and make it opt-in for a time period. It's still the model of 1 parent context → 1 child sync but with a buffer to process many contexts at once. I think that fits your use case, but there's also many more possibilities that I would like to support eventually.

Thank you for linking those Edgar! Even if not implemented, they are interesting to review and might provide some insights.