Merge multiple documents to an array in ultra pipeline

@JensDeveloper - Ultra tasks consume incoming documents. I believe what you want to do is POST data to the Ultra task, not pass it as URL query parameters. See the Ultra Pipeline best practices documentation final bullet point in regards to your situation with trying to parse the pipeline parameters:

• You cannot change Pipeline parameters at run time. These parameters are specified in the Pipeline itself or when the task is created. They are set at pipeline startup, which only happens on initiation, not per document.

It should be noted that I’m recommending that you push the creation of the array back to the source database because you are trying to prevent calling a child pipeline, which is the recommended pattern for aggregating data in an Ultra pipeline. This is required because Ultra is a streaming service of one record in, one record out. So any type of function that aggregates data (including Sort, Group By *, etc) will be stuck waiting on end of upstream input which will never come because of the nature of the data flow in Ultra. This is also why your Script snap isn’t working the way you were expecting - it’s waiting on the end of input, which never comes in an Ultra pipeline.

Hi @koryknick

Thank you for the answer. I aslo found my own mistake in ultra tasks you just get the url by using the variable $path_info in a mapper. I used a pipeline parameter _PATH_INFO but that only works in triggered task.

Regards

Jens

Hello @thilkip,

You can try and use Group by N snap. Set the group size to 0 and this is what you get:

Regards,
Bojan

Hi @thilkip ,
Use Pipeline Execute with the option Reuse executions to process documents set to disabled. Inside this pipeline put your input data and use group by n. Something like this:

Use this script, it groups all inputs into single array:

execute : function () {
   this.log.info("Executing Transform Script");
    var groupByN = new ArrayList();
    var wrapper = {};
    while (this.input.hasNext()) {
        try {
            // Read the next input document, store it a new LinkedHashMap, and write this as an output document.
            // We must use a serializable Java type liked LinkedHashMap for each output instead of a native
            // JavaScript object so that downstream Snaps like Copy can process it correctly.
            var inDoc = this.input.next();

            groupByN.add(inDoc);
            wrapper = {"Data": groupByN};
        }
        catch (err) {
            var errDoc = new LinkedHashMap();
            errDoc.put("error", err);
            this.log.error(err);
            this.error.write(errDoc);
        }
    }
    this.output.write(wrapper);
    this.log.info("Script executed");
}

I tested it with ultra task and it seems it works.

@thilkip - what is generating the input documents?

SQL Server 2016 (13.x) and later has a “FOR JSON” clause:

You should be able to return the result set exactly how you want.

I don’t have a DB2 instance to test with at the moment, but it appears you could do the same thing with different syntax:

Hi again @thilkip,

This task of yours was very intriguing, so I dedicated quite sometime experimenting and I found a solution that works in development mode, with snaps supported by Ultra Task. Here is the pipeline:

Documents_to_Array_Format.slp (7.3 KB)

Here is the final result:

I am pretty curious about this one, can you please send feedback on the pipeline’s behaviour?
In addition I am sending you a Python script, that is also tested and works. I know that you allready tried with a script but I am offering some options. Thanks,

Doc_to_Array.txt (1.6 KB)

Best Regards,
Bojan