Best practice for dynamic 'De-serialize From File' in a Producer/Consumer pattern

[zoomingrocket]

I am implementing a Producer/Consumer pattern to optimize a fast REST-to-Slow Oracle data movement.

Producer Pipeline: Fetches paginated data from a REST API and uses the Serialize to File transform to dump pages to local .ser files.
Consumer Pipeline: Needs to pick up these files via Get File Names, use De-serialize from File, and push to an Oracle DB.

The Problem:
The De-serialize from File transform requires a static filename or a variable; it does not currently accept a filename from an incoming field in the stream. Since I am processing multiple files, I cannot easily use a single variable in a single pipeline execution without wrapping it in a complex loop/workflow.

Questions for the Community:
Is Metadata Injection (MDI) the recommended way to dynamically pass the filename into the De-serialize transform, or is there a more lightweight approach?
In a high-throughput scenario, is there a performance penalty to using MDI for every file vs. a workflow loop?
Are there plans to add "Accept filename from field" to the De-serialize transform to match the behavior of the Text File Input transform?

I have tested this pattern by writing out raw REST API data via text output but this uses a lot of disk space so pivoted to serialization as it offers much better performance & compression, just stuck with last mile to Deserialize Files 🤷‍♂️

Thank for you the inputs in advance.

[dmainou]

Hey variables is fine but I think you got this not quite right.
2 distinct processes running in parallel

Process 1
REST call -> landing/filename_timestamp.json

Process 2
get current time as batch_id
/landing/filename_.*.json -> /archive/batch_id/filename_.*.json
/archive/batch_id/filename_.*.json -> oracle

super simple and scalable.
Add error handling + backfill and you are done.

[zoomingrocket]

@dmainou - Yup, we have that working without serialization but raw flat files per API page are 6-8MB vs a serialized binary file is just 350kb! We need to scale this to 100s of similar pipelines so trying to be conservative about transient disk space needs. Also the performance with serialized file vs raw text is considerably better.

[dmainou]

ok then

Process 1

Pipeline1
P_DATA_SOURCE = customers (parameter)
get current time
create filename (e.g. P_DATA_SOURCE + "_" + timestamp+ ".ser"
pipeline executor passing filename as P_FILENAME (parameter)

Pipeline 2
P_FILENAME
REST call -> landing/P_FILENAME

Process 2

Pipeline 1
get current time as batch_id
/landing/filename_..json -> /archive/batch_id/filename_..json
get list of files in /archive/batch_id/filename_.*.json
pipeline executor passing filename as P_FILENAME (parameter)

Pipeline 2
P_FILENAME -> oracle

[zoomingrocket]

@dmainou - Thanks for the tip on the parameter way of sending a variable, which helped resolve it via pipeline executor. We are managing in a single workflow, performing both source fetch and parallel push to the database. This helps decouple the source fetch loop, which is faster v/s target Oracle DB Push, establishing a healthy, high-performance pipeline!

[dmainou]

Hey,
From the screenshot.
Parallel execution needs parallel design and understanding of consequences and interactions.
My 2 cents is that is far easier to just run in sequence or separate workflows.

Annnd, yes, loosely coupled logic keeps things simple.

[dmainou]

and just to nit pick.

take a look at these
https://hop.apache.org//manual/latest/pipeline/transforms/pipeline-logging.html
https://hop.apache.org//manual/latest/pipeline/transforms/workflow-logging.html
and look at the samples.

Far better logging capability than that logging you are executing.
Both have their place, and if you want to scale to hundreds of api calls and run 24*7 then you want to use the above to write your metrics elsewhere. Likely a place where you can throw observability (e.g. prometheus) and governance (open metadata/collibra) over it.

[zoomingrocket]

@dmainou - Great suggestion, I will explore it further! The current job metrics are just capturing & printing total rows processed vs inserted vs updated vs deleted in log for our stakeholder KPIs. Since we run all pipelines via Airflow + Hop run, we push all of it via Airflow Open Telemetry export and aggregate the data for KPIs.

We did try direct Hop run auto instrumentation using Open Telemetry Java Agent, but since Hop uses custom logging it was did not capture anything, but I will definitely explore dispatching Pipeline metrics to Otel.
Thanks!