How to log current batch number and increment it in pipeline executor when using row grouping?

[ossDataEngineer]

Hello Hop Community,

I have a usecase where source system gives me records in huge batches. The target system accepts smaller batch size, so I need to chunk src records into smaller batches with the chunk/batch size = max allowed by target system.

Eg. SRC gives 1k records, TGT accepts 200 records, so I need to create 5 batches as folllowing

tgt_batch_1.json, tgt_batch_2.json, ....etc.

I was able to create smaller batches of the data using pipeline executor row grouping as explained in samples/loops/child-generate-1000-rows.hpl but now I want to maintain counter variable so I can keep track of the iteration.

How can I do this?

[dave-csc]

Your best bet would be using some transforms before entering the Pipeline Executor loop:

• Add sequence: use it to add a sequence number on each row, starting from zero
• Calculator: divide the sequence number by your batch size (200), floor it and force it to an Integer, then add 1 if necessary

The output of the Calculator will be your batch number. And you can use it to group the rows for the Pipeline Executor, too ;)

Hope this helps :)

[dave-csc]

Simply using Get variables and Set variables in the pipeline 3 should do the trick:

• use Get variables at the start of your pipeline, and retrieve both variables FAILED_RECORD_COUNT and SUCCESS_RECORD_COUNT in some field
• in the success path, be sure to reduce the stream to a single row, then use a Calculator to increment the field SUCCESS_RECORD_COUNT, and set the new value to the variable with Set variables
• do the same in the failure path, using field and variable FAILED_RECORD_COUNT

Anyway, to detect if a pipeline had succeeded, I would also take a look at the Execution results in pipeline 2...

[ossDataEngineer]

@dave-csc The problem I see is that every pipeline executor instance gets variable value = 0 which was set in the parent workflow. The state of variable is not updated across different pipeline executor instances.

For example:

Parent WF set SUCCESS_RECORD_COUNT=0 and FAILED_RECORD_COUNT=0,
Repeat Pipeline creates 5 instances of pipelines because of row-grouping
When the 3rd pipeline runs, all the Get Variables inside it get 0 value. So if the

batch 1 had SUCCESS_RECORD_COUNT=150 and FAILED_RECORD_COUNT=50
batch 2 had SUCCESS_RECORD_COUNT=200 and FAILED_RECORD_COUNT=0
batch 3 had SUCCESS_RECORD_COUNT=150 and FAILED_RECORD_COUNT=50
batch 4 had SUCCESS_RECORD_COUNT=200 and FAILED_RECORD_COUNT=0
batch 5 had SUCCESS_RECORD_COUNT=70 and FAILED_RECORD_COUNT=130

The final count is always being set to last batch failed/success counts (i.e 70 and 130) and not adding across all batches.

[ossDataEngineer]

child_pipeline.hpl.txt
grandchild_pipeline.hpl.txt
parent_workflow.hwf.txt

[dave-csc]

Hi @ossDataEngineer,

I had a quick look of your pipelines/workflow.

First, you are not implementing the CURRENT_BATCH_COUNT as described in the previous post. The value of the counter must be input from the child pipeline, not calculated into the grandchild one. If you need to use that as a "variable", it actually has to be passed as a parameter in the grandchild pipeline.

Regarding the SUCCESS_ and FAILED_RECORD_COUNT, the behaviour is coherent with the fact that you're not returning the variable to the workflow after setting it, so it is reset at each iteration. To get what you need, I would proceed like this:

• in the grandchild pipeline, add a Group by transform at the end of the success and the failure stream, to count the rows processed by each one
• join the two grouped streams with a Join Rows (cartesian product) (don't worry, they should be a single row each...) and send the result to the output with Copy rows to result
• in the child pipeline, grab the Result rows to the next transform: you can either log directly every batch result with a Write to log on the fields received from the grandchild, or place another Group by in between to calculate the cumulative sum of the success and failed number fields
• if you still need to have those values in some variables, use another Group by after logging in the child pipeline (take the sums of the success and failed fields, or the maximum values of the cumulative sums if you calculated those). Then you can finally use Set variables, and those can be used in the parent workflow

Hope this helps :)

[ossDataEngineer]

Thankyou so much @dave-csc for detailed solution, it helped. Appreciate your time on this!