Apache Nifi use cases: Dataflow to Ingestion sample data PDF Free Download
1 / 18/18
100%
Apache Ni use cases: Dataow to Ingeson
sample data
2
Table of Contents
1 Disclaimer ....................................................................................................................... 3
2 Overview of Apache Nifi Interface ..................................................................................... 3
3 Pipeline organization ........................................................................................................ 5
3.1 Example Structure ................................................................................................... 5
3.2 Use case Hierarchical Process Groups : ................................................................... 5
4 Use case implementation: Ingest Presences table from CDR Database.............................. 6
4.1 Short description ..................................................................................................... 6
4.2 Controller services ................................................................................................... 6
4.3 Data Pipeline Flow ................................................................................................... 9
4.3.1 Flow Overview ...................................................................................................... 9
4.3.2 Flow explanation ................................................................................................ 10
3
1 Disclaimer
NSO may intend to process Call Detail Records (CDRs) obtained from the telecommunicaons
operator.
To demonstrate the use case for mobile data (which may originate from an operator), sample mobile
data has been loaded into a PostgreSQL database. These sample datasets are available as open-
source resources and can be accessed at the following repository:
hps://github.com/alisdair/nodobo-release/tree/master.
2 Overview of Apache Ni Interface
The NiFi User Interface (UI) facilitates the creaon, visualizaon, modicaon, monitoring, and
management of automated dataows. It comprises mulple segments, each serving a disnct
funcon within the applicaon. This secon presents visual representaons of the interface and
outlines its key components.
When a user navigates to the UI for the rst me, a blank canvas is provided on which a dataow can
be built:
The Components Toolbar, located in the upper le poron of the interface, provides a selecon of
elements that can be dragged onto the canvas to construct dataows. A detailed explanaon of each
component is provided in the Building a DataFlow secon.
Posioned below the Components Toolbar, the Status Bar displays real-me system metrics, including
acve processing threads, data volume within the ow, and the state of various process groups and
processors (e.g., transming, running, disabled). In clustered deployments, it also indicates the
number of nodes and their connecvity status. The Status Bar is periodically refreshed, with the
latest update mestamp provided.
On the le side of the screen, the Operate Palee oers controls for user to oversee workows and
for administrators to manage user access and congure system properes, including resource
allocaon for the applicaon.
To the right of the canvas, the Search Funcon enables ecient component discovery by name, type,
idener, or conguraon properes.
4
Adjacent to Search, the Global Menu provides opons (see the screenshot below) for modifying
exisng components within the dataow environment.
The user interface includes navigaon features to facilitate ecient movement across the canvas. The
Navigate Palee enables users to pan and zoom, while the Bird’s Eye View provides an overview of
the dataow, allowing navigaon across extensive secons. Addionally, breadcrumbs at the boom
of the screen display the navigaon path within Process Groups, indicang the hierarchy and depth
of the ow. Each breadcrumb serves as a link, enabling users to seamlessly return to previous levels
within the dataow structure.
5
Certain walkthrough resources
1
, which are further elaborated upon, provide a more comprehensive
overview of the NiFi interface.
3 Pipeline organizaon
Upon logging in, the user is directed to the main canvas, commonly referred to as the NiFi Flow. The
NiFi Flow can be conceptually compared to a home directory on a computer, where organizing data
pipelines is similar to structuring work within folders and subfolders.
In Apache NiFi, the Process Group component funcons similarly to a directory, serving as a logical
container for organizing workows. Eecve pipeline management in Apache NiFi relies primarily on
the structured arrangement of Process Groups and their nested sub-Process Groups.
To ensure an opmal organizaon, it is recommended to structure pipelines into logically dened
Process Groups, categorizing them by department, dataset, and data segments (e.g., tables). This
approach enhances clarity, maintainability, and operaonal eciency in dataow management.
3.1 Example Structure
➔ Nif main low
➔ Department 1 (process group):
➔ Sub-ow (process group) for Dataset A.
➔ Sub-ow (process group) for each table if needed
➔ Sub-ow (process group) for Dataset B.
➔ Sub-ow (process group) for each table if needed
➔ Department 2 (process group):
➔ Sub-ow (process group) for Dataset C.
➔ Sub-ow (process group) for each table if needed
➔ Sub-ow (process group) for Dataset D.
➔ Sub-ow (process group) for each table if needed
3.2 Use case Hierarchical Process Groups :
In our case, we will ingest the "Presences" table from the "CDR" dataset for the Department of
Demographic and Social Stascs.
Below is the hierarchical structure of the project:
1
hps://www.youtube.com/watch?v=nisWXIRXyq0
6
4 Use case implementaon: Ingest Presences table from CDR Database
4.1 Short descripon
This case outlines the process of ingesng data from a PostgreSQL database and implemenng
paroning mechanisms within the data ow. Specically, it details the procedure for incorporang
paron elds into the content of the owle, dynamically generang parons based on owle
content aributes, and eciently storing the structured data in a MinIO bucket using Apache NiFi.
4.2 Controller services
Within the framework of Apache NiFi, a Controller Service constutes a shared service designed to
facilitate the reuse of conguraons and resources across mulple processors, reporng tasks, or
other services within a NiFi data ow. This mechanism enables the centralized management and
governance of conguraons, which is essenal to the interoperability of various system
components.
7
For the present use case, a single Controller Service is required, namely the DBCPConneconPool,
which shall be designated as PG_CDRDB_ConneconPool. The DBCPConneconPool serves as a
Controller Service in Apache NiFi, managing a pool of database connecons ulizing Apache
Commons DBCP (Database Connecon Pooling). This conguraon opmizes database interacons
by enabling NiFi processors to establish ecient and sustainable communicaon with relaonal
databases, thereby migang the need for repeated connecon establishment and terminaon.
Consequently, this approach enhances system performance and ensures opmal resource ulizaon.
To create a DBCPConneconPool, follow these steps:
a. Open Controller Services Panel
❑ Click on the Gear Icon () at the top-right corner of the “Table: Presences” process group
Conguraon.
❑ Navigate to Controller Services.
b. Add a New Controller Service
❑ Click the "+" buon to add a new Controller Service.
❑ Search for DBCPConneconPool and select it.
❑ Click "Add".
c. Congure the DBCPConneconPool
❑ Click on the CView Conguraon () buon to congure the connecon pool.
8
❑ Change de default “Name” on « SETTINGS » tab
❑ Enter the following details on « PROPERTIES » tab :
Property
Value
Database Connecon URL
For MySQL: jdbc:mysql://<HOST>:<PORT>/<DATABASE>
For PostgreSQL: jdbc:postgresql://<HOST>:<PORT>/<DATABASE>
For SQL Server: jdbc:sqlserver://<HOST>:<PORT>;databaseName=<DATABASE>
Database Driver Class Name
MySQL: com.mysql.cj.jdbc.Driver
PostgreSQL: org.postgresql.Driver
SQL Server: com.microso.sqlserver.jdbc.SQLServerDriver
Database Driver Locaon(s)
/opt/ni/drivers
Database User
your_username
Password
your_password
Max Total Connecons
10
Validaon Query
SELECT 1
d. Enable the Connecon Pool
❑ Click Apply to save the sengs.
❑ Click on the Enable () buon to acvate the service.
9
4.3 Data Pipeline Flow
4.3.1 Flow Overview
When ingesng a database for the rst me, it is common pracce to account for the historical data
already present before scheduling the ingeson of new data at a well-dened frequency. NSO may
process mobile data on a monthly basis.
The data ingeson workow eecvely demonstrates this approach by retrieving both historical and
newly generated data from the "Présences" table each month. The data from the "Présences" table
will be stored as paroned Parquet les, with each paron represenng the volume of data
collected for a given month.
Outlined below are the steps and processors ulized in construcng the data ingeson workow for
the "Présences" table:
10
1- GenerateFlowFile: Generates two owles containing the aributes startDate and endDate,
represenng the inial and current data load parameters.
2- UpdateAribute: Computes and assigns the aribute monthDuraon to the owle using a
predened constant milliSeconds, established in the preceding GenerateFlowFile processor.
3- RouteOnAribute: Evaluates whether the computed startDate does not exceed endDate and
routes the owle accordingly.
4- UpdateAribute: Updates the startDate aribute in the owle to the subsequent month.
5- UpdateAribute: Incorporates dataset paroning aributes into the owle to enable
structured data management.
6- GenerateTableFetch: Constructs an SQL query using startDate and endDate as ltering
criteria.
7- ExecuteSQL: Executes the dynamically generated SQL query to retrieve the required dataset.
8- ConvertAvroToParquet: Converts the extracted Avro le format into Parquet to enhance
storage eciency and processing performance.
9- PutS3Object: Transfers and stores the Parquet le in MinIO to ensure accessibility and long-
term data management.
4.3.2 Flow explanaon
4.3.2.1 GenerateFlowFile (s):
The pipeline begins with two GenerateFlowFile processors, each congured to handle dierent
aspects of data ingeson
1. Conguraon for Historical Data
Three new properes are added to the processor:
• startDate: The starng date for retrieving historical data, formaed as YYYY-MM, compable
with the loaded database.
• endDate: The ending date for the historical data, formaed as YYYY-MM.
• milliSeconds: A constant that will be used later to calculate the duraon of one month.
11
2. Conguraon for New Data
Similarly, three new properes are added to the processor:
• startDate: The starng date is automacally determined using Apache NiFi’s expression
language
2
.
startDate = ${now():format('yyyy-MM')}
• endDate: The ending date is automacally determined using Apache NiFi’s expression
language.
endDate = ${now():format('yyyy-MM')}
• milliSeconds: A constant that will be used later to calculate the duraon of one month.
These two processors are linked to the next stage of the pipeline through the Funnel component,
ensuring a streamlined ow of data.
A Funnel ( )is a component used to merge mulple data ows into a single stream or distribute
data from one source to mulple desnaons.
4.3.2.2 UpdateAribute
2
hps://ni.apache.org/docs/ni-docs/html/expression-language-guide.html
12
This processor is supplied with input from the Funnel ( ) component. In the event of a
successful execution, its output is directed to the RouteOnAttribute processor.
Congure the processor as shown below
Add a new property monthDuraon. This property will later be used as an incremental value in the
loop that iterates through the months from startDate to endDate.
Using Apache NiFi’s expression language, the monthDuraon is calculated in milliseconds based on
the following formula:
monthDuraon = 31 days × 24 hours × 60 minutes × 60 seconds × 1000 milliseconds
monthDuration =
${milliSeconds:toNumber():multiply(31):multiply(24):multiply(60):multiply(60)}
This calculaon ensures that the monthly duraon is consistently represented in milliseconds for
further processing within the data pipeline.
4.3.2.3 RouteOnAribute
Congure the processor as below
13
Within the RouteOnAribute processor in Apache NiFi, a new property notLast has been introduced
using Apache Ni expression language, dened as follows :
notLast = ${startDate:toDate("yyyy-MM"):le(${endDate:toDate("yyyy-MM")})}
This property serves the purpose of evaluang whether the startDate is less than or equal to the
endDate. If startDate is not greater than endDate, the following acons are performed:
• Using the UpdateAribute (1) processor, dataset aributes are updated to facilitate the
storage of data as paroned Parquet les in MinIO.
• Using the UpdateAribute (2) processor, the startDate is incremented to proceed with the
next data ingeson cycle.
4.3.2.4 UpdateAribute:
Congure the processor as shown below
startDate = ${startDate:toDate("yyyy-
MM"):toNumber():plus(${monthDuration}):format("yyyy-MM")}
This property serves the funcon of dynamically updang the startDate aribute by incremenng it
based on a dened duraon (monthDuraon).
This processor, along with the preceding RouteOnAribute processor, forms a loop that iterates
through each month from startDate to endDate, ensuring the sequenal processing of data over
me.
14
Aer incremenng the startDate property, this processor forwards its output (FlowFile) as a new
input to the previously idened Funnel component.
4.3.2.5 UpdateAribute:
This processor receives as input the FlowFile from the RouteOnAribute processor and updates the
dataset informaon required to query the database and store the data in MinIO.
Congure the processor as shown below
The following aributes must be incorporated to facilitate data paroning:
• datasetMonth: The month relevant to the query. This value is automacally extracted from
the startDate aribute using Apache NiFi's expression language.
datasetMonth = ${startDate:toDate("yyyy-MM"):format("MM")}
• datasetName: The name of the dataset.
• datasetYear: The year corresponding to the dataset month, dynamically derived from the
startDate aribute.
15
datasetYear = ${startDate:toDate("yyyy-MM"):format("yyyy")}
• department: The department responsible for the dataset.
• tableName: The name of the database table.
These aributes ensure proper organizaon and retrieval of data within the pipeline.
4.3.2.6 GenerateTableFetch
This processor ulizes the aributes dened by the preceding UpdateAribute processor to
construct the database query.
Congure this processor as shown below
The following core properes are modied:
• Database Connecon Pooling Service = PG_CDRDB_ConneconPool: The
DBCPConneconPool service, previously created as a prerequisite, which provides a
connecon to the database.
• Database Type = PostgreSQL: The database type, which can be PostgreSQL, MySQL, MS SQL
2012+, etc.
• Table Name = Presences: The name of the target database table.
16
• Addional WHERE Clause: A lter applied to the dataset to retrieve records for a specic
month = date_part('year', created_at) = ${datasetYear} AND
date_part('month', created_at) = ${datasetMonth}
4.3.2.7 ExecuteSQL:
Congure this processor as shown below
The database connecon service must be specied by modifying the Database Connecon Pool
Service property.
The ExecuteSQL processor in Apache NiFi produces data in Avro format
4.3.2.8 ConvertAvroToParquet.
Since the objecve is to store the data in Parquet format, this processor is responsible for converng
the data from Avro format to Parquet format, ensuring compability with the targeted storage and
processing requirements.
Congure the processor as shown below
17
The selected compression type is SNAPPY to priorize query performance
3
4.3.2.9 PutS3Object:
Once the data for the month has been collected and transformed into Parquet format, this processor
will store it in MinIO for further processing and analysis.
Congure the processor as shown below
3
https://dev.to/alexmercedcoder/all-about-parquet-part-05-compression-techniques-in-parquet-4bcb
18
The list of core properes to be modied is as follows:
• Object Key: The absolute path of the le.
Object Key =
${department}/${datasetName}/${tableName}/${datasetYear}/${datasetMonth}/da
ta.parquet
• Bucket: The name of the bucket.
• Access Key ID: The access key generated by MinIO.
• Secret Access Key: The secret key generated by MinIO.
• Endpoint Override URL: The API URL of MinIO (port 9000).
