MODULE -3 Reading and writing files
CONTENT
[Link] and writing csv files using
pandas dataframes
[Link] json with python
[Link] with csv and json in nifi
[Link] pipelines with apache airflow
APACHE AIRFLOW – DATA PIPELINE
What Airflow Is
•Apache Airflow is a workflow orchestration
tool
•Used to schedule, monitor, and manage data
pipelines
•Pipelines are defined using Python code, not
GUI
How Airflow Pipelines Work (python)
[Link] a DAG in Python
[Link] work into tasks
[Link] dependencies between tasks
[Link] runs tasks in order
[Link] shows success or failure
Term Meaning
DAG Directed Acyclic Graph – the pipeline
Task One unit of work
Operator Defines what a task does
Scheduler Triggers tasks based on time/dependencies
Executor Runs tasks
Web UI Monitors pipeline status
Use Case
Every day:
•Read a CSV file
•Clean the data
•Load it into a database
Pipeline Steps
Step Task What Happens
1 Extract Read [Link]
2 Transform Remove null values
3 Load Insert data into database
Extract_CSV → Transform_Data → Load_To_DB
How Tasks Are Defined (Simplified)
•Extract task
Reads CSV from storage
•Transform task
Cleans and formats data
•Load task
Inserts data into database
Each task runs only after the previous one succeeds.
1. DAG (Pipeline Definition)
•dag_id → Unique pipeline name
•start_date → When pipeline becomes active
•schedule_interval="@daily" → Runs once every day
•catchup=False → No backfilling of past runs
DAG = the full pipeline
2. Tasks (Work Units)
Task Role
extract_csv Reads CSV file
transform_data Cleans and formats data
load_to_database Loads data into DB
3. Operators
•PythonOperator runs Python functions
•One operator = one task
•Airflow has many operators (Bash, SQL, Spark, etc.)
•Scheduler triggers the DAG
•Extract task runs
•If successful → Transform runs
•If successful → Load runs
•Status visible in Airflow UI
WORKING WITH CSV AND JSON IN NIFI
APACHE NIFI
•Nifi is a flow automation tool, like Apache Airflow.
•But it was built to work via GUI instead of
programming.
UNDERSTANDING FLOWFILES AND PROCESSORS
•FlowFiles are the basic data units in Apache NiFi.
Each FlowFile has:
•Content → the actual data (for example, rows in a file)
•Attributes → metadata such as filename, size, or timestamps
•This distinction matters because NiFi operations act either on
content or on attributes.
•In this scenario:
•The content represents the data rows
•The filename attribute is used to filter FlowFiles and route different data to different
processors
•Processors are NiFi components that perform operations or transformations.
•They work like functions in a program
•They receive FlowFiles, process them, and pass them to the next processor
Step What Happens Example
NiFi receives data from a Read a CSV file from a local
1. Data Ingestion
source folder
Data is wrapped as a FlowFile CSV rows = content, filename =
2. FlowFile Creation
(content + attributes) attribute
FlowFiles are routed using Files ending with [Link] go to
3. Routing / Filtering
attributes Processor A
Processors transform or act on
4. Processing Convert CSV to JSON
data
FlowFiles are passed to next Send processed data to
5. Transfer
processor database
Store data in HDFS / S3 /
6. Data Output Data is delivered to destination
MySQL
Simple Example Flow (GUI-based)
Use case: Process two CSV files using filename
[Link] Processor
1. Reads files from /input/
2. Example files:
[Link]
[Link]
[Link] Processor
1. Condition:
[Link] contains "sales"
[Link] contains "customers"
[Link]
1. Sales flow → ConvertRecord (CSV → JSON)
2. Customer flow → PutDatabaseRecord
[Link]
1. Sales data stored in JSON format
2. Customer data stored in database
Use case: Read CSV → clean → split → convert → store in database
[Link]
1. Input file: [Link]
[Link]
1. Replace empty values with 0
[Link]
1. Uses CSVReader
2. Splits records into batches of 1,000
[Link] Reader (CSVReader)
1. Reads CSV using schema
2. Schema fields: order_id, product, amount, date
[Link] Writer (JsonRecordSetWriter)
1. Converts records into JSON
[Link] Service
[Link]
2. Stores database URL, username, password
[Link] Access
1. Schema Registry or inline Avro schema
2. Ensures correct data types before insert
[Link]
1. Uses Record Reader + Writer
2. Inserts data into sales_table
three CSV files in one input directory:
•[Link]
•[Link]
•[Link]
Each file must be processed differently.
Flow Explanation (Step-by-Step)
1. GetFile
•Reads all CSV files together
•Creates one FlowFile per CSV
•Attributes include: filename, path, size
Output FlowFiles:
[Link], [Link], [Link]
2. ReplaceText
•Performs content cleanup
•Example:
• Replace empty values with 0
• Standardize date format
Applied before routing so all files are clean.
3. RouteOnAttribute
•Routes files using filename
Routing rules:
•filename contains "sales" → Sales Flow
•filename contains "customers" → Customer Flow
•filename contains "products" → Product Flow
This is the key step for handling multiple CSVs in one pipeline.
4. SplitRecord
•Used in each routed flow
•Splits CSV into smaller record batches (e.g., 1,000 rows)
Why it matters:
Large CSVs cannot be efficiently loaded into databases as one file.
5. Record Reader (CSVReader)
•Reads rows using a schema
•Converts raw text into structured records
Example schema fields:
•Sales: order_id, amount, date
•Customers: customer_id, name, city
•Products: product_id, category, price
6. Record Writer (JsonRecordSetWriter)
•Converts records into JSON format
•Keeps schema consistency
Used internally by record-based processors.
7. PutDatabaseRecord
•Inserts records into database tables
Target tables:
•[Link] → sales_table
•[Link] → customer_table
•[Link] → product_table
Uses Controller Service (DBCPConnectionPool) for database connection.
PROBLEM
We’ll get three different files in the csv format. Inject the data into the
Postgres database
CREATE COMPANIES TABLE
CREATE TABLE companies (
id bigint NOT NULL,
name text NOT NULL,
image_url text,
created_at timestamp without time zone NOT NULL,
updated_at timestamp without time zone NOT NULL
);
CREATE CAMPAIGNS TABLE
CREATE TABLE campaigns (
id bigint NOT NULL,
company_id bigint NOT NULL,
name text NOT NULL,
cost_model text NOT NULL,
state text NOT NULL,
monthly_budget bigint,
blacklisted_site_urls text[],
created_at timestamp without time zone NOT NULL,
updated_at timestamp without time zone NOT NULL
);
CREATE ADS TABLE
CREATE TABLE ads (
id bigint NOT NULL,
company_id bigint NOT NULL,
campaign_id bigint NOT NULL,
name text NOT NULL,
image_url text,
target_url text,
impressions_count bigint DEFAULT 0,
clicks_count bigint DEFAULT 0,
created_at timestamp without time zone NOT NULL,
updated_at timestamp without time zone NOT NULL
);
GETTING THE DATA
Download and move to a known directory the following files:
[Link]
[Link]
[Link]
READING THE FILES
To Make anything in Nifi, we need processors.
To get a File and read its content, we need the GetFileProcessor.
To add a Processor drag and drop to the canvas the upper left icon
GETFILE PROCESSOR
To Make anything in
Nifi, we need
processors.
To get a File and read
its content, we need
the
GetFileProcessor.
To add a Processor
drag and drop to the
canvas the upper left
icon
CONFIGURING PROPERTIES
To change the processor behavior we can change its properties, to do so, click with right
mouse button in the processor and choose the option configure.
The GetFileProcessor has the following properties
INPUT DIRECTORY
The properties that are in bold are required for the Processor work.
In the pic you can see the Input Directory and the File
Filter properties, they are the key properties.
The input directory is the path to the directory where the files you’ve
download are located
FILTERS
In file filter you can either put the file name like [Link] but we want to get
all three files, that’s why the value is the following
[^\.]*.csv
It just saying that all files that ends with .csv must be a target
ROUTING THE FLOWFILES
Now that we can get the three files we need to send them to different
processors flows because each one has a different structure, and each
one is going to a different table in the database.
To do that, we need a processor called RouteOnAttribute
Choose Route to Property Name, that says to processor that each
FlowFile that match a condition needs to be redirect independently,
If you provide 3 different conditions, there will be 3 path available to
send the flowfile.
ROUTING
AFTER ROUTING
CONNECTING THE TWO PROCESSORS
Now we need to tell nifi that the flowfiles got in the GetFile Processor
need to be passed to the RouteOnAttribute Processor. We do this using
the Relationship concept. It is just a link between two processors.
To connect both, hover over the GetFile and you are going to seee a
arrow, drag and drop the arrow to the RouteOnAttribute processors.
For this case there is just Success relationship, meaning that all
FlowFiles that were successfully processed, are going to be passed via
this relationship
CONNECTING TWO PROCESSORS
ADDING HEADERS TO THE CSV FILE
our csv files do not have a header, meaning the the first row is already the data itself, so
there’is no way of knowing each colum.
Just edit the file and add the following headers above the first line.
USING REPLACETEXT PROCESSOR
To achieve this using a processor, we will use the ReplaceText
processor. Since we need to route FlowFiles to three different
paths, we will use three ReplaceText processors—one for each
path.
Let’s start by configuring one of them:
Drag three ReplaceText processors onto the canvas.
Select any one of them and open the Properties tab to begin
configuration.
To avoid confusion, it's a good practice to rename each processor
using the Settings tab.
CONFIGURING REPLACETEXT PROPERTIES
In the Properties of the ReplaceText processor, we need to modify the replacement strategy.
Our goal is to add a header line above the first line of the file. For this example, we will
configure it for the [Link] file.
Follow these steps:
•Set Replacement Strategy to Prepend.
•In the Replacement Value, copy and paste the desired header line.
•Change Evaluation Mode to Entire Text.
(If left as default, the header will be added before every line instead of just once.)
SPLITTING THE RECORDS
Now we need to split the rows to insert into the databse.
Again we need three different instances for the same processor for each path.
The processor we’re going to use is the SplitRecord processor, pretty straightforward
In the properties we have just three properties.
Let’s focus on the Records Per Split, in our case we want one row each time, so we need to set
the property to be 1.
RECORD READER AND RECORD WRITER
In the Record Reader and Record Writer we need to pass Controllers.
It is just helpers to provide to the controller the right interpretation for something.
In this case we need to provide controller that will tell the processor that the flowfile
coming is from a csv source
To provide the Reader, click on the Record Reader property value, click on the drop
down and then click on the Create New Service
CONFIGURE CONTROLLER SERVICES
CHANGING THE PROPERTIES
SCHEME ACCESS
In schema access, we choose the infer schema option, so the controller will know all
the columns.
The other important property is treat first line as header, that must be set to true, so it
won’t be handled as data itself.
Click in “ok”, and finally we need to enable the controller
INSERTING INTO DATABASE
Finally we can get each record (or row in our case) and insert into the database.
use the PutDatabaseRecord processor.