Basics Tutorial

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

In this section, we will create a Processor that reads the CSV file and filters for new stations.
We know from the test run what format the data is in and must specify the subsequent Processor for processing.

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

The QueryRecord Processor uses SQL queries to select content from incoming FlowFiles. To do this, the data must be in a format that NiFi can interpret as records—the CSV format is one such example.

You can then use SQL to filter, transform, or aggregate the data.

The query results are passed along as FlowFiles via a custom-defined relation.

First, we’ll 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 file containing information about all new train stations. We could then transfer this compressed XML file to external systems using various Processors. For example, we could use the PutEmail Processor to send the file as an attachment or use PutSFTP to store it on a file server. In many cases, we pass the content as a file.

We can specify the name of this file by setting the “filename” attribute, which is then read by the Processors. Even though we’re only downloading the content via the IGUASU user interface in this tutorial, the name of the download file is also set by this attribute.
With the UpdateAttribute Processor, we can create and modify any attributes of the FlowFiles.

We add an UpdateAttribute Processor to the Diagram and connect it to the * “neue_alle” relation* from the QueryRecord Processor.

New attributes are created using the UpdateAttribute Processor by defining a dynamic property.
The name of this dynamic property corresponds to the name of the attribute to be created (or modified). The content corresponds to the (new) value.

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

To do this, we use Expression Language, which allows us to include dynamic functions in expressions for properties.

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

The UpdateAttribute Processor only modifies the defined attributes of the FlowFile, not the content. We will forward the resulting FlowFile—now with the updated “filename”—to the next Processor.

In this step, we compress the FlowFile. To do this, we use the CompressContent Processor. We need to add the Processor to the Diagram and simply create the relationship 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 in the next step or saved for backup purposes. We simulate this step by using IGUASU’s analysis tools to download the content, unpack the file, and finally view the XML.
To do this, we route the FlowFiles through a Funnel element so we can 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 specifically “capturing” FlowFiles via the user interface for manual inspection without having them processed by another Processor.

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

Click on this waiting FlowFile to view the FlowFile and an overview of the queue on this relation on the right-hand side.

Using the icon in the content area, you can download the FlowFile content.

When we download the file, we should see that the name is “Stations_[yyyy-MM-dd].xml.gz”—as we specified earlier. 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 train stations as XML.

First, we need a Processor that receives and forwards the HTTP requests. The HandleHttpRequest Processor is suitable for this purpose.
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 StandardHttpContextMap Controller Service, which ensures that the HTTP session is maintained between the Processors.

As we saw in Part 1, we create a new StandardHttpContextMap Controller Service, but this time within the current Process Group for 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, since the Controller Services of a Process Group are also available to all sub-Process Groups. However, we’re avoiding that here because we want to use the Controller Service exclusively within this Process Group.

When creating this Controller Service, it’s a good idea to prefix the name with “REST” to avoid confusion with other StandardHttpContextMap Controller Services.

We set the optional Allowed Paths property to “/validate”.
As a result, the Processor will only accept requests with this path in the URL. All requests that do not go to localhost:3333/validate are automatically responded to with a 404 (Not Found) status code.

Since we are providing a REST service to which data is to be sent for validation, the client should use the “POST” HTTP method when sending the data. Therefore, we uncheck the “Allow GET,” “Allow PUT,” and “Allow DELETE” settings so that only “Allow POST” remains enabled.

To respond to the requests, we also need two additional HandleHttpResponse Processors.
The first one should return a 200 status code to the client as a success message upon successful validation. The other, in turn, should notify the client of an error with a 400 status code.

In a real-world scenario, it would be appropriate to handle the different error cases (invalid JSON provided, JSON schema validation failed) differently and to provide additional information about the error. For this tutorial, we will limit ourselves to returning a 400 status code for both error cases.

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

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

In the next step, we want to check the received JSON data for the correct format. The ValidateJson Processor is well-suited for this purpose, as it can validate JSON based on a JSON schema.

A JSON schema is itself a JSON file that specifies the structure and content of another JSON—in other words, it defines a schema. The ValidateJson Processor checks whether our JSON meets the requirements of the JSON schema.

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

This schema must be entered into 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 returns response code 400.

The resulting flow should look like this:

Our REST Service for verifying the data, implemented as a subordinate Process Group, is now complete and ready to be started.

In this part, we want to retain only the station number (EVA_NR as NR) and the station name (NAME) from the CSV data for the stations to provide a clearer overview. We can use the QueryRecord Processor we created earlier to remove all other information and, as before, convert the data to XML format at the same time.
To do this, we create a new Dynamic Property named “new” and set the appropriate SQL statement as its value.

Both “neue” and “neue_alle” are now outputs of the QueryRecord Processor. To verify the functionality, we can run a test using the existing CSV data (see Isolated Process Execution (Continued)).

We want both Part 1 and Part 2 to be executed every time the flow runs. Therefore, we will not modify Part 1, but rather 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 we just created. This processor thus forwards two different FlowFiles to two separate outputs, which are processed independently of one another.

In addition to the station number and name, we now want to add the current date to each entry. This is to indicate that the information is up to date. We’ll 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. Since we receive our data in XML format from QueryRecord, we need an XMLReader.

As in Part 1, we must add this as a Service to the Process Group and set the “Expect Records as Array” option to * “true”* in the configuration, since the data arrives as an array in the XML (see “Result content” at Test run of the QueryRecord Processor).

For the Record Writer, we’ll use a JsonRecordSetWriter Processor with default settings. This converts the records from XML to JSON format, since we ultimately want to send a JSON response to the REST interface.

Next, we’ll add a Dynamic Property using the button. We’ll use the expression for the current date again and add it 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 (Setting File Names).)

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

We now forward the FlowFile containing the JSON to a SplitRecord Processor. Using this Processor, we will split the list of stations into individual elements so that the REST service can be called for each station.

Since the data arrives in JSON format, we need a JSONTreeReader as the 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. With this Processor, we call the REST Service to validate the JSON data and pass the FlowFile content—that is, the JSON containing the entry for a train station—as the body.

The flow now looks as follows:

Right-click in the Diagram and select Start to launch all Processors.

The top InvokeHTTP Processor sends an HTTP request to retrieve the CSV file. As soon as it receives the file via the response, the FlowFile is passed on to the QueryRecord Processor.

Here, the flow splits into two branches.
On the left, an XML file containing all new stations and all available information is passed to the UpdateAttribute Processor via the “neue_alle” relation. The UpdateAttribute Processor sets the “filename” attribute, which determines the name of the file that will be output later.

The FlowFile, which contains both content and attributes, is passed on to the CompressContent Processor, where the XML is converted into a gzip-compressed data stream. This data stream ultimately 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 passes an XML file via the “new” relation that also contains data on the new train stations—though only the names and numbers. The UpdateRecord Processor then adds additional information to each of these entries: today’s date.

At the same time, it converts the data into the JSON format expected by the REST endpoint. This JSON containing station entries is then sent to the SplitRecord Processor, where each entry is individually packaged into a FlowFile, so that multiple FlowFiles—rather than just one—now leave the Processor.

Each of these FlowFiles passes individually through the “splits” response into the InvokeHTTP Processor, which sends a request containing the JSON from the content to the REST Service we implemented for each FlowFile.

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