Basic Tutorial

First we will create an InvokeHTTP Processor.
The InvokeHTTP Processor can execute any HTTP requests. We download the CSV file with a GET request to the corresponding URL. The file is translated internally into a FlowFile and can be forwarded to other Processors.

In this section, we will create a Processor that receives the CSV file and filters it for new stations.
We know from the test execution what format the data is in and need to determine the subsequent Processor for processing.

One of the Processors we can use for our concern is the QueryRecord Processor.

The QueryRecord Processor selects the content of incoming FlowFiles via SQL queries. To do this, the data must be available in a format that NiFi can interpret as records - the CSV format is one of these.

You can then filter, transform or aggregate via the SQL.

The results of the queries are forwarded as FlowFiles along a self-defined relation.

First we will add the Processor to the Diagram and connect it to the Response Relation of the InvokeHTTP Processor.

Next, we need to configure the QueryRecord Processor so that it handles the incoming data as desired.

Our goal for part 1 is to save a compressed XML with information about all new stations. We could then transfer this compressed XML as a file to external systems via various Processors. For example, we could send the file as an attachment using the PutEmail Processor or store it on a file server using PutSFTP. In many cases, we transfer the content as a file.

We can determine the name of this file by setting the "filename" attribute, which is then read by the Processors. Even if we only download the content via the IGUASU user interface in this tutorial, the name of the download file is also set by this attribute.
We can use the UpdateAttribute Processor to create and change any attributes of the FlowFiles.

We add an UpdateAttribute Processor to the Diagram and connect it to the "new_all" relation from the QueryRecord Processor.

Creating new attributes via the UpdateAttribute Processor is done by creating a dynamic property.
The name of this dynamic property then corresponds to the name of the attribute that is to be created (or changed). The content then corresponds to the (new) value.

We want to set the content to the current date. As the date will change daily, we cannot enter the string statically, but must generate it dynamically to match the day.

To do this, we use the Expression Language, with which we can enter dynamic functions in expressions for properties.

The expression ${now():format('yyyy-MM-dd')} would result in a current date such as "2020-11-19". As we want to save the file as XML, we can append the static ".xml" to the expression and prefix it with "Stations_": Stations_${now():format('yyyy-MM-dd')}.xml.
This then results in "Stations_2020-11-19.xml".

The UpdateAttribute Processor only changes the defined attributes of the FlowFile, not the content. We will forward the resulting FlowFile with the now adjusted "filename" to the next Processor.

In this step, we compress the FlowFile. We use the CompressContent Processor for this. We need to add the Processor to the Diagram and only create the relation from the UpdateAttribute.

In the settings, we should:

We forward the compressed FlowFile to the last Processor.

We can imagine that the compressed XML file will be forwarded to an external system or saved for backup in the next step. We simulate this step by using the IGUASU analysis tools to download the content, unpack the file and finally view the XML.
To do this, we run the FlowFiles into a Funnel element to view them later.

A Funnel is a simple element in IGUASU. It is used to collect FlowFiles from one or more incoming connections, but terminates the data flow at this point. This is useful for terminating branches of a flow or for 'catching' FlowFiles specifically for manual inspection via the user interface without another Processor processing them.

By right-clicking in the Diagram and selecting Start, we can start all Processors in the Process Group. If we have done everything correctly, we should see after a short time that a FlowFile has run through our Processors and is now waiting in front of the Funnels.

We click on this waiting FlowFile to see the FlowFile and an overview of the queue on this relation on the right-hand side.

We can download the content of the content via the icon in the content area.

When we download the file, we should see that the name is "Stations_[yyyy-MM-dd].xml.gz" - as previously determined by us. The file extension ".gz" indicates a file compressed with gzip. If we unzip the file and open the resulting XML in an editor, we can now view the new stations as XML.

First, we need a Processor that receives and forwards the HTTP requests. The HandleHttpRequest Processor is suitable for this.
In the configuration, we set the listening port to "3333". This is the port on which the Processor will listen for requests.

We also need to create a new standard HttpContextMap Controller Service, which ensures that the HTTP session between the Processors is maintained.

As known from part 1, we create a new standard HttpContextMap Controller Service, but this time within the current Process Group validation via REST.
We could also create the Controller Service in the parent Process Group "Deutsche Bahn Tutorial" or even in the Root Process Group, as the Controller Services of a Process Group are also available to all Sub-Process Groups. However, we avoid this here, as we only want to use the Controller Service in this Process Group.

When creating this Controller Service, it is advisable to prefix the name with "REST" to avoid confusion with another standard HttpContextMap Controller Services.

We fill the optional Allowed Paths property with "/validate".
As a result, the Processor only accepts requests with this path in the URL. All requests that do not go to localhost:3333/validate are automatically answered with a 404 (Not Found) code.

Because we offer a REST Service to which data should be sent for verification, the client should use the "POST" HTTP method when sending the data. We therefore uncheck the "Allow GET", "Allow PUT" and "Allow DELETE" settings so that only "Allow POST" is activated.

To answer the requests, we also need two additional HandleHttpResponse Processors.
The first should return a code 200 to the client as a success message if validation is successful. The other, in turn, should inform the client with a code 400 if an error occurs.

In a real scenario, it would be appropriate to answer the different error cases (no valid JSON transferred, JSON schema validation failed) differently and also provide further information about the error. For this tutorial, we will limit ourselves to the response with the code 400 for both error cases.

We create two HandleHttpResponse Processors and set our "REST StandardHttpContextMap" Controller Service in the HTTP Context Map field in both cases.

We set the HTTP Status Code field to 200 and 400 respectively.

In the next step, we want to check the JSON data received for the correct format. The ValidateJson Processor, which can validate JSONs using JSON schemas, is suitable for this.

A JSON schema is itself a JSON that specifies the structure and content of another JSON, i.e. defines a schema. The ValidateJson Processor checks whether our JSON meets the specifications of the JSON schema or not.

With the following JSON schema, we define a JSON object that represents an entry for a station. It has exactly three entries:

This schema must be entered in the "JSON Schema" property. Otherwise, the Processor’s default settings can be retained.

The resulting configuration of the ValidateJson Processor should ultimately look like this:

The valid entries are sent along the "valid" relation to the HandleHttpResponse Processor with the successful response code 200, while in the event of a validation error, the other HandleHttpResponse Processor sends back the response code 400.

In the end, the following flow should result:

Our REST Service implemented as a subordinate Process Group for checking the data is now ready and can be started.

In this part, we want to keep only the station number (EVA_NR as NR) and the station name (NAME) from the CSV data of the stations for a better overview. We can use the previously created QueryRecord Processor to remove all other information and at the same time, as before, transfer the data into an XML format.
To do this, we create a new Dynamic Property with the name "new" and set the appropriate SQL statement as the value.

Both "new" and "new_all" are now outputs of the QueryRecord Processor. To check the function, we can do a test execution with the existing CSV data (see Isolated process execution (continuation)).

We want both part 1 and part 2 to be run every time the flow is executed. Therefore, we will not change part 1, but let both parts exist and run in parallel.

To do this, we connect our next new Processor, the UpdateRecord Processor, to the "new" output of the QueryRecord Processor that has just been created. This thus forwards two different FlowFiles to two different outputs, which are handled independently of each other.

We now want to add the current date to each entry in addition to the number and name of the station. This is to show that the information corresponds to today’s position. We use the UpdateRecord Processor for this.

As with the QueryRecord Processor in part 1, we need to specify a Record Reader Controller Service and a Record Writer Controller Service. As we receive our data in XML format from QueryRecord, we need an XMLReader.

As in the first part, we need to add this as a Service to the Process Group and set the Expect Records as Array setting to "true" in the configuration, as the data arrives as an array in XML (see "Result content" at Test execution of the QueryRecord Processor).

We will use a JsonRecordSetWriter Processor with default settings for the Record Writer. This converts the stations from XML to JSON format, as we ultimately want to send a JSON to the REST interface.

We then add a dynamic property using the button. We use the expression for the current date again and add this to each record. The name of the dynamic property specifies the name of the additional field in the record.
(This procedure is similar to the previous step with the UpdateAttribute Processor in the first part of the tutorial, see 1.3 UpdateAttribute Processor (set file name).)

During execution, the stations are converted from XML to JSON and provided with the current date after the name "CREATED":

We now forward the FlowFile with the JSON to a SplitRecord Processor. We will use this Processor to split the list of stations into individual elements so that one call to the REST Service can be made per station.

As the data arrives in JSON format, we need a JSONTreeReader as a Record Reader, which we must again create as a Controller Service.
The default settings can be retained.

For the Record Writer, we can use the JsonRecordSetWriter that we previously created for the UpdateRecord Processor.

Each FlowFile from the SplitRecord Processor is now forwarded to another InvokeHTTP Processor. We use this Processor to call the REST Service to check the JSON data and transfer the FlowFile content, i.e. the JSON with the entry for a station, as the body.

The flow now looks as follows:

By right-clicking in the Diagram and selecting Start, we start all Processors.

The upper InvokeHTTP Processor sends an HTTP request to receive the CSV file. As soon as it receives this with the response, the FlowFile is forwarded to the QueryRecord Processor.

Here the flow splits into two branches.
Links, an XML of all new stations with all existing information is passed to the UpdateAttribute Processor via the "new_all" relation. The UpdateAttribute Processor is used to set the "filename" attribute, which determines the name of the file output later.

The FlowFile comprising content and attributes is passed on to the CompressContent Processor, where the XML is turned into a data stream compressed with gzip. This finally ends up in the relation to the Funnel, where we can view and download the content.

On the other branch of the flow, the QueryRecord Processor also passes an XML with data on the new stations, but only the name and number, via the "new" relation. The UpdateRecord Processor then adds further information to each of these entries, namely today’s date.

At the same time, it converts the data into the JSON format expected by the REST endpoint. This JSON with entries for stations then goes into the SplitRecord Processor, where each entry is individually packed into a FlowFile, so that now several FlowFiles leave the Processor instead of one.

Each of these FlowFiles runs individually via the "splits" response into the InvokeHTTP Processor, which sends a request with the JSON from the content to the REST Service we have implemented for each of the FlowFiles.

Each FlowFile with valid JSON data produces a response from the REST Service with status code 200 and is routed to the "Response" relation accordingly.