EXP - 2.

Installing and configuring Elasticsearch, Kibana, PostgreSQL, pgAdmin 4

To install and configure Elasticsearch, Kibana, PostgreSQL, and pgAdmin 4, follow these
steps for each tool. I will provide instructions for a typical Linux system (Ubuntu), but the
steps are similar for other platforms.

1. Installing and Configuring Elasticsearch

Elasticsearch is a distributed search and analytics engine. It stores data in the form of indices
and allows real-time searching and querying.

a. Install Elasticsearch

1. Install prerequisites:

bash
CopyEdit
sudo apt update
sudo apt install apt-transport-https openjdk-11-jdk wget

2. Add Elasticsearch GPG key and repository:

bash
CopyEdit
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo
apt-key add -
sudo sh -c 'echo "deb https://artifacts.elastic.co/packages/8.x/apt
stable main" > /etc/apt/sources.list.d/elastic-8.x.list'

3. Install Elasticsearch:

bash
CopyEdit
sudo apt update
sudo apt install elasticsearch

4. Configure Elasticsearch (optional):

o The configuration file is located at /etc/elasticsearch/elasticsearch.yml.
o Modify settings like network.host and http.port if you need to expose
Elasticsearch to other machines. Example:

yaml
CopyEdit
network.host: 0.0.0.0
http.port: 9200

5. Start and enable Elasticsearch:

bash
CopyEdit
sudo systemctl start elasticsearch
sudo systemctl enable elasticsearch

6. Verify Elasticsearch is running:

 bash
CopyEdit
curl -X GET "localhost:9200/"

2. Installing and Configuring Kibana

Kibana is the visualization and management UI for Elasticsearch.

a. Install Kibana

1. Install Kibana using apt:

bash
CopyEdit
sudo apt update
sudo apt install kibana

2. Configure Kibana:
o The configuration file for Kibana is /etc/kibana/kibana.yml.
o Modify the server.host to bind Kibana to the desired interface. Example:

yaml
CopyEdit
server.host: "0.0.0.0"
elasticsearch.hosts: ["http://localhost:9200"]

3. Start and enable Kibana:

bash
CopyEdit
sudo systemctl start kibana
sudo systemctl enable kibana

4. Access Kibana:
o Kibana should now be available at http://<your_server_IP>:5601.

3. Installing and Configuring PostgreSQL

PostgreSQL is a powerful, open-source relational database system.

a. Install PostgreSQL

1. Install PostgreSQL:

bash
CopyEdit
sudo apt update
sudo apt install postgresql postgresql-contrib

2. Check PostgreSQL status:

bash
CopyEdit
sudo systemctl status postgresql
 3. Start PostgreSQL:

bash
CopyEdit
sudo systemctl start postgresql

4. Enable PostgreSQL to start on boot:

bash
CopyEdit
sudo systemctl enable postgresql

5. Switch to PostgreSQL user:

bash
CopyEdit
sudo -i -u postgres

6. Access PostgreSQL prompt:

bash
CopyEdit
psql

7. Create a database:

sql
CopyEdit
CREATE DATABASE my_database;

8. Create a user:

sql
CopyEdit
CREATE USER my_user WITH ENCRYPTED PASSWORD 'my_password';

9. Grant privileges:

sql
CopyEdit
GRANT ALL PRIVILEGES ON DATABASE my_database TO my_user;

10. Exit PostgreSQL:

sql
CopyEdit
\q

4. Installing and Configuring pgAdmin 4

pgAdmin is a popular graphical tool for managing PostgreSQL databases.

a. Install pgAdmin 4

1. Add pgAdmin repository:

bash
CopyEdit
sudo sh -c 'echo "deb
https://ftp.postgresql.org/pub/pgadmin/pgadmin4/deb/ubuntu/ focal
pgadmin4" > /etc/apt/sources.list.d/pgadmin4.list'

2. Add the GPG key:

bash
CopyEdit
curl https://www.pgadmin.org/static/packages_pgadmin_org.pub | sudo
apt-key add -

3. Install pgAdmin 4:

bash
CopyEdit
sudo apt update
sudo apt install pgadmin4

4. Configure pgAdmin 4:
o After installation, pgAdmin 4 is accessible via a web browser.
o To configure, you can run the following command:

bash
CopyEdit
sudo /usr/pgadmin4/bin/setup-web.sh

o This will ask for an email and password for pgAdmin's web interface.

5. Access pgAdmin 4:
o After configuration, open a browser and go to http://localhost/pgadmin4.
o Log in using the email and password you set up.

EXP-3. Reading and Writing files

a. Reading and writing files in Python

b. Processing files in Airflow

c. NiFi processors for handling files

d. Reading and writing data to databases in Python

e. Databases in Airflow f. Database processors in NiFi

a. Reading and Writing Files in Python
Python provides several built-in libraries to read and write files, such as open(), csv, and
json. Here's an example of basic file handling in Python:

Reading a file:

python

Copy

Edit

# Reading a text file

with open('example.txt', 'r') as file:

content = file.read()

print(content)

Writing to a file:

python

Copy

Edit

# Writing to a text file

with open('output.txt', 'w') as file:

file.write('Hello, this is a test!')

Reading and writing CSV files:

python

Copy

Edit

import csv

# Reading CSV

with open('input.csv', 'r') as file:

reader = csv.reader(file)
 for row in reader:

print(row)

# Writing CSV

with open('output.csv', 'w', newline='') as file:

writer = csv.writer(file)

writer.writerow(['Name', 'Age'])

writer.writerow(['Alice', 25])

Reading and writing JSON files:

python

Copy

Edit

import json

# Reading JSON

with open('data.json', 'r') as file:

data = json.load(file)

print(data)

# Writing JSON

with open('output.json', 'w') as file:

json.dump({"name": "Alice", "age": 25}, file)

b. Processing Files in Airflow

Apache Airflow is an open-source workflow orchestration platform that can be used to

automate and schedule tasks, including reading and writing files. You can use Python scripts
or specific operators in Airflow to interact with files.

Airflow Operators for File Handling:

PythonOperator: Execute Python code within your Airflow DAG, including file handling tasks.
BashOperator: You can also use the BashOperator to execute shell commands for file
handling.

Example of reading and writing files in Airflow:

python

Copy

Edit

from airflow import DAG

from airflow.operators.python import PythonOperator

from datetime import datetime

def read_file():

with open('example.txt', 'r') as file:

print(file.read())

def write_file():

with open('output.txt', 'w') as file:

file.write('Airflow is awesome!')

with DAG('file_processing_dag', start_date=datetime(2025, 2, 11),

schedule_interval='@daily') as dag:

read_task = PythonOperator(task_id='read_file', python_callable=read_file)

write_task = PythonOperator(task_id='write_file', python_callable=write_file)

read_task >> write_task

c. NiFi Processors for Handling Files

Apache NiFi is a data integration tool that provides a wide range of processors to handle file-
based operations. Some of the key processors for handling files include:

GetFile: Reads files from a directory.

PutFile: Writes files to a directory.

FetchFile: Retrieves a file based on the file path.

SplitText: Splits a text file into smaller chunks (by lines, for example).
PutDatabaseRecord: Writes data to a database (related to database handling).

PutHDFS: Writes files to a Hadoop HDFS system.

Example Flow for Reading and Writing Files in NiFi:

Use the GetFile processor to read files from a directory.

Process the files (e.g., split, transform, or filter).

Use PutFile to write the processed data to another directory or system.

d. Reading and Writing Data to Databases in Python

Python provides libraries like sqlite3, pymysql, psycopg2 (for PostgreSQL), and SQLAlchemy
for interacting with databases.

Example of Reading and Writing Data with SQLite in Python:

python

Copy

Edit

import sqlite3

# Connect to SQLite database (or create it)

conn = sqlite3.connect('example.db')

cursor = conn.cursor()

# Create table

cursor.execute('''CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT,
age INTEGER)''')

# Insert data

cursor.execute("INSERT INTO users (name, age) VALUES ('Alice', 30)")

# Commit and close connection

conn.commit()

# Reading data

cursor.execute("SELECT * FROM users")

rows = cursor.fetchall()
for row in rows:

print(row)

# Close the connection

conn.close()

e. Databases in Airflow

Airflow provides operators and hooks to interact with various databases, such as MySQL,
PostgreSQL, SQLite, etc.

PostgresOperator: Allows executing SQL statements on PostgreSQL databases.

MySqlOperator: Allows executing SQL statements on MySQL databases.

PythonOperator: Can be used to execute custom Python code to interact with databases.

Example of a Database Task in Airflow:

python

Copy

Edit

from airflow import DAG

from airflow.providers.postgres.operators.postgres import PostgresOperator

from datetime import datetime

with DAG('database_processing_dag', start_date=datetime(2025, 2, 11),

schedule_interval='@daily') as dag:

create_table = PostgresOperator(

task_id='create_table',

postgres_conn_id='postgres_default',

sql='''

CREATE TABLE IF NOT EXISTS users (

id SERIAL PRIMARY KEY,

name VARCHAR(100),

age INTEGER
 );

create_table

f. Database Processors in NiFi

Apache NiFi offers processors that allow you to interact with databases, some of the most
commonly used ones are:

ExecuteSQL: Executes a SQL query on a relational database and returns the result.

PutSQL: Inserts or updates records in a relational database.

QueryDatabaseTable: Retrieves data from a database table.

GenerateTableFetch: Used to generate SQL queries to retrieve records from a database

table.

Example NiFi Flow for Database Operations:

ExecuteSQL: To query data from a database.

PutSQL: To insert data into a database.

QueryDatabaseTable: To extract data from a database table periodically.

EXP- 4. Working with Databases

a. Inserting and extracting relational data in Python

b. Inserting and extracting NoSQL database data in Python

c. Building database pipelines in Airflow

d. Building database pipelines in NiFi

Working with databases in Python and integrating them into automated workflows can be
broken down into a few main tasks:

a. Inserting and Extracting Relational Data in Python

Relational databases like MySQL, PostgreSQL, SQLite, and others are commonly accessed
through Python using connectors or ORMs (Object-Relational Mappers).

1. Using sqlite3 (for SQLite):

SQLite is a lightweight, file-based relational database system.

python

Copy

Edit

import sqlite3

# Connect to a database (or create it if it doesn't exist)

conn = sqlite3.connect('example.db')

cursor = conn.cursor()

# Create a table

cursor.execute('''CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT,
age INTEGER)''')

# Insert data

cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ('Alice', 30))

conn.commit()

# Extract data

cursor.execute("SELECT * FROM users")

rows = cursor.fetchall()

for row in rows:

print(row)

# Close the connection

conn.close()

2. Using psycopg2 (for PostgreSQL):

You can interact with PostgreSQL databases using libraries like psycopg2:

python

Copy

Edit
import psycopg2

# Connect to PostgreSQL

conn = psycopg2.connect(dbname="your_db", user="your_user",

password="your_password", host="localhost")

cursor = conn.cursor()

# Insert data

cursor.execute("INSERT INTO users (name, age) VALUES (%s, %s)", ('Bob', 25))

conn.commit()

# Extract data

cursor.execute("SELECT * FROM users")

rows = cursor.fetchall()

for row in rows:

print(row)

# Close the connection

conn.close()

3. Using SQLAlchemy (ORM for relational databases):

SQLAlchemy is a popular ORM that abstracts raw SQL queries and allows you to work with
database objects more naturally.

python

Copy

Edit

from sqlalchemy import create_engine, Column, Integer, String

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import sessionmaker

# Define a base class

Base = declarative_base()

# Define a model
class User(Base):

__tablename__ = 'users'

id = Column(Integer, primary_key=True)

name = Column(String)

age = Column(Integer)

# Create an engine and session

engine = create_engine('sqlite:///example.db')

Session = sessionmaker(bind=engine)

session = Session()

# Create tables

Base.metadata.create_all(engine)

# Insert data

new_user = User(name="Charlie", age=35)

session.add(new_user)

session.commit()

# Extract data

users = session.query(User).all()

for user in users:

print(f"{user.name}, {user.age}")

session.close()

b. Inserting and Extracting NoSQL Database Data in Python

NoSQL databases, like MongoDB, are non-relational and are accessed differently.

1. Using pymongo (for MongoDB):

python

Copy

Edit
from pymongo import MongoClient

# Connect to MongoDB

client = MongoClient("mongodb://localhost:27017")

db = client["testdb"]

collection = db["users"]

# Insert data

collection.insert_one({"name": "Alice", "age": 30})

# Extract data

users = collection.find({"age": {"$gte": 25}})

for user in users:

print(user)

# Close connection

client.close()

2. Using cassandra-driver (for Apache Cassandra):

python

Copy

Edit

from cassandra.cluster import Cluster

# Connect to Cassandra

cluster = Cluster(['127.0.0.1'])

session = cluster.connect('testkeyspace')

# Insert data

session.execute("INSERT INTO users (id, name, age) VALUES (uuid(), 'Bob', 25)")

# Extract data

rows = session.execute("SELECT * FROM users")

for row in rows:

 print(row)

# Close connection

cluster.shutdown()

c. Building Database Pipelines in Airflow

Apache Airflow is used to automate workflows and can be used to build database pipelines,
whether it's for ETL processes or scheduled database operations.

Example of a simple pipeline that extracts data from one database and loads it into another:

Install Airflow and necessary dependencies for your database:

bash

Copy

Edit

pip install apache-airflow

pip install apache-airflow-providers-postgres

pip install psycopg2

Define a simple DAG (Directed Acyclic Graph):

python

Copy

Edit

from airflow import DAG

from airflow.providers.postgres.operators.postgres import PostgresOperator

from airflow.providers.postgres.hooks.postgres import PostgresHook

from datetime import datetime

def extract_data():

# Extract data from PostgreSQL

pg_hook = PostgresHook(postgres_conn_id='source_postgres')

query = "SELECT * FROM source_table"

 records = pg_hook.get_records(query)

return records

def load_data():

# Load data into PostgreSQL

pg_hook = PostgresHook(postgres_conn_id='target_postgres')

records = extract_data()

for record in records:

pg_hook.run(f"INSERT INTO target_table (column1, column2) VALUES {record}")

default_args = {

'owner': 'airflow',

'start_date': datetime(2025, 2, 11),

'retries': 1,

with DAG('database_pipeline', default_args=default_args, schedule_interval='@daily') as

dag:

extract = PostgresOperator(

task_id='extract_data',

postgres_conn_id='source_postgres',

sql="SELECT * FROM source_table",

load = PostgresOperator(

task_id='load_data',

postgres_conn_id='target_postgres',

sql="INSERT INTO target_table VALUES (...)", # Modify based on your query

extract >> load

d. Building Database Pipelines in NiFi

Apache NiFi is a tool for automating the flow of data between systems. To create a database
pipeline in NiFi, you use processors that interact with databases.

Create a NiFi Flow:

Use processors like ExecuteSQL, PutSQL, or QueryDatabaseTable.

ExecuteSQL allows you to extract data from a database.

PutSQL allows you to insert data into a database.

Example of Creating a Flow:

Extract data: Use the ExecuteSQL processor to run a query and extract data.

Transform data: You can use processors like TransformXml, EvaluateJsonPath, or

ExecuteScript to transform the data.

Load data: Use PutSQL to insert the transformed data into the destination database.

The general flow in NiFi looks like this:

ExecuteSQL (extract data) → PutSQL (insert data into another database).

EXP- 5. Cleaning, Transforming and Enriching Data

a. Performing exploratory data analysis in Python

b. Handling common data issues using pandas

c. Cleaning data using Airflow

5. Cleaning, Transforming, and Enriching Data

This section focuses on techniques for performing exploratory data analysis (EDA), handling
common data issues, and cleaning data in a more automated way using Airflow. Let’s break
it down:

a. Performing Exploratory Data Analysis (EDA) in Python

Exploratory Data Analysis (EDA) is the first step in analyzing the data to understand its
structure, identify any patterns, and uncover relationships. Python, particularly with libraries
like pandas, matplotlib, and seaborn, makes it straightforward to perform EDA.

Here are the common steps involved in EDA:

Load the Data: Load your dataset into a DataFrame using pandas.

python

Copy

Edit

import pandas as pd

data = pd.read_csv("data.csv")

Check Data Structure:

python

Copy

Edit

data.info() # Check data types and missing values

data.describe() # Summary statistics

Visualize Data:

Histograms, box plots, or scatter plots can help visualize distributions and relationships.

python

Copy

Edit

import matplotlib.pyplot as plt

import seaborn as sns

sns.histplot(data['column_name']) # Histogram for distribution

plt.show()

sns.boxplot(x='column_name', data=data) # Box plot for outliers

plt.show()

Correlation Analysis:

Check correlations between numeric variables.

python

Copy

Edit

corr_matrix = data.corr()

sns.heatmap(corr_matrix, annot=True, cmap="coolwarm")

plt.show()

Identify Missing Values:

python

Copy

Edit

data.isnull().sum() # Count missing values in each column

b. Handling Common Data Issues Using Pandas

Data issues like missing values, duplicates, and incorrect data types can be handled
effectively using pandas.

Handling Missing Data:

Drop rows with missing values:

python

Copy

Edit

data.dropna(inplace=True)

Fill missing values:

python

Copy

Edit

data.fillna(value='Unknown', inplace=True) # Replace with a value

data['column_name'].fillna(data['column_name'].mean(), inplace=True) # Replace with
mean value

Remove Duplicates:

python

Copy

Edit

data.drop_duplicates(inplace=True) # Remove duplicate rows

Change Data Types:

python

Copy

Edit

data['column_name'] = data['column_name'].astype('float64')

Handling Categorical Data:

Convert categorical variables into numeric:

python

Copy

Edit

data['category_column'] = data['category_column'].astype('category')

data['category_column'] = data['category_column'].cat.codes # Converts categories to

numeric codes

Renaming Columns:

python

Copy

Edit

data.rename(columns={'old_name': 'new_name'}, inplace=True)

c. Cleaning Data Using Airflow

Airflow is a powerful tool for automating and scheduling data pipelines. Cleaning data using
Airflow generally involves creating a Directed Acyclic Graph (DAG) to automate the steps of
extracting, transforming, and loading (ETL) the data.

Setting Up Airflow DAG:

A DAG in Airflow represents the workflow. Here's a basic structure of an Airflow DAG for
data cleaning:

python

Copy

Edit

from airflow import DAG

from airflow.operators.python_operator import PythonOperator

from datetime import datetime

def clean_data():

import pandas as pd

data = pd.read_csv('data.csv')

data = data.dropna() # Example of cleaning data

data.to_csv('cleaned_data.csv', index=False)

dag = DAG('data_cleaning', description='Data cleaning pipeline',

schedule_interval='@daily', start_date=datetime(2025, 2, 11), catchup=False)

clean_task = PythonOperator(task_id='clean_data', python_callable=clean_data, dag=dag)

Task Dependencies:

In Airflow, tasks can be chained together. For example, you can first extract data, clean it,
then load it.

python

Copy

Editextract_task >> clean_task >> load_task # Chaining tasks

Automating the Workflow:

The above DAG is set to run daily, and you can schedule the pipeline to run as needed.

To enable logging, you can integrate logging with tasks.

python

Copy

Edit

def clean_data():

import logging

logging.info("Cleaning the data...")

data = pd.read_csv('data.csv')

data = data.dropna() # Clean operation

data.to_csv('cleaned_data.csv', index=False)

logging.info("Data cleaning completed.")

Error Handling in Airflow:

Tasks can also have retries and notifications if they fail.

python

Copy

Edit

clean_task = PythonOperator(

task_id='clean_data',

python_callable=clean_data,

retries=3, # Retry 3 times on failure

email_on_failure=True, # Send email on failure

dag=dag

EXP- 6. Building the Data Pipeline

Building a data pipeline involves the process of collecting, processing, storing, and analyzing
data efficiently. Here's a step-by-step guide to building a data pipeline:

1. Define the Pipeline Objectives

Before starting, identify the objectives of the pipeline:

 What data are you collecting?

 What is the source of the data?
 What kind of analysis will be done?
 What are the outcomes expected from the pipeline?
 Understand any performance or scalability constraints.

2. Select Data Sources

Choose the sources of data to be collected. These sources can include:

 Databases (SQL, NoSQL)

 APIs (RESTful, SOAP)
 Files (CSV, JSON, XML)
 Real-time streams (e.g., Kafka)
 Third-party services (social media feeds, etc.)

3. Data Ingestion

Data ingestion is the process of collecting raw data and feeding it into the pipeline.
Depending on the nature of your data, the ingestion method will vary:

 Batch Processing: Collecting data at set intervals (e.g., daily, hourly).

 Stream Processing: Continuous flow of data, such as live data streams.

Tools you can use for ingestion:

 Apache Kafka: For stream-based ingestion.

 Apache Nifi: For both batch and stream ingestion.
 AWS Glue / Google Dataflow: Managed cloud services for data ingestion.

4. Data Processing

After ingestion, the next step is transforming the data into a useful format. This can include:

 Cleaning: Removing duplicates, handling missing values, and correcting errors.

 Transformation: Normalizing, aggregating, or enriching the data.
 Enrichment: Combining data from different sources to add value.
 Filtering: Reducing noise by excluding irrelevant data.

Tools for data processing:

 Apache Spark: Distributed processing for large datasets.

  Apache Flink: Real-time stream processing.
 ETL (Extract, Transform, Load) Tools: Talend, Airflow, Informatica.

5. Data Storage

The transformed data needs to be stored for further analysis or reporting. The storage can be
structured (e.g., relational databases) or unstructured (e.g., data lakes).

 Relational Databases (SQL): MySQL, PostgreSQL, etc.

 Data Warehouses: Amazon Redshift, Google BigQuery, Snowflake.
 Data Lakes: Hadoop, AWS S3, Azure Blob Storage, Google Cloud Storage.

6. Data Analysis

After storing the data, the next step is to analyze it to derive insights:

 Use SQL or NoSQL queries for basic analysis.

 Implement advanced analytics or machine learning using platforms like Apache
Spark, Google Cloud AI, or AWS SageMaker.
 Utilize business intelligence tools like Tableau, Power BI, or Looker for
visualizing the data.

7. Data Presentation

After analysis, presenting data to stakeholders in a user-friendly way is essential. This can
include dashboards, reports, or interactive visualizations.

 Dashboards: Tools like Grafana, Power BI, or Tableau can be used to build
interactive visualizations.
 Automated Reports: Use scripting or reporting tools to generate regular reports.

8. Orchestration

Orchestrating the data pipeline ensures smooth, scheduled, and automated execution of each
stage in the pipeline. It handles dependencies and retries for failed steps. Common tools
include:

 Apache Airflow: For scheduling and managing workflow dependencies.

 Luigi: Another Python-based tool for pipeline orchestration.

9. Monitoring & Logging

Monitoring ensures that the pipeline is running smoothly. This includes:

 Checking performance metrics (e.g., processing time, error rates).

 Setting up alerts for failures or delays.
 Keeping logs of data processes for troubleshooting.

Tools for monitoring and logging:

  Prometheus & Grafana: For monitoring.
 Elasticsearch, Logstash, Kibana (ELK Stack): For log aggregation and analysis.

10. Pipeline Maintenance and Optimization

Data pipelines require ongoing maintenance to ensure they remain efficient, scalable, and
responsive to new data sources or business needs:

 Scaling: Add resources or optimize processing for increased data volume.

 Error Handling: Implement graceful error handling for data failures.
 Cost Optimization: Minimize storage and compute costs by optimizing the pipeline.

EXP- 7. Building a Kibana Dash Board

Building a Kibana dashboard involves a series of steps to visualize and interact with your
data stored in Elasticsearch. Here’s a step-by-step guide on how to create one:

Step 1: Set up Elasticsearch and Kibana

Before you can build a dashboard, you need to have:

 Elasticsearch running and configured to index data.

 Kibana running, connected to Elasticsearch, and accessible through a web interface.

Step 2: Import Data into Elasticsearch

 You need to have data indexed into Elasticsearch. This data could come from logs,
application metrics, etc.
 Use tools like Logstash, Beats, or directly index data into Elasticsearch through
APIs.

Step 3: Access Kibana

 Open your browser and go to Kibana's web interface (typically,

http://localhost:5601).
 Log in if required.

Step 4: Define an Index Pattern

In Kibana, you need an Index Pattern that corresponds to the data in Elasticsearch.

1. Go to Kibana > Management > Stack Management > Index Patterns.

2. Click Create Index Pattern.
3. Choose the index or wildcard that matches the data in Elasticsearch.
4. Define the timestamp field (if applicable).
5. Click Create Index Pattern.

Step 5: Explore Your Data (Discover)

  Go to Discover in Kibana to see the raw data from your Elasticsearch index.
 You can use filters, sort, and search your data here to get an overview before building
visualizations.

Step 6: Create Visualizations

Kibana provides various visualization types such as bar charts, line graphs, pie charts, maps,
etc.

1. Go to Visualize Library or Visualize.

2. Click Create Visualization and choose the type of visualization (e.g., bar chart, line
graph, pie chart, etc.).
3. Select the index pattern for the data you want to visualize.
4. Define the aggregations (metrics like sum, average, count) and splits (field to break
the data into).
5. Customize the visualization with color schemes, legends, labels, and axis settings.

Once the visualization is complete, click Save and give it a name.

Step 7: Create the Dashboard

Now that you have your visualizations ready, you can create a dashboard.

1. Go to Dashboard in Kibana.
2. Click Create new Dashboard.
3. Click Add and choose the visualizations you’ve created.
4. You can resize, arrange, and configure each visualization’s display.
5. You can also use Filters and Time Ranges to customize the data shown on the
dashboard.
6. After arranging the visualizations to your preference, click Save and provide a name
for the dashboard.

Step 8: Share or Embed the Dashboard

 You can share your dashboard by clicking on the Share button in the top right corner.
 Kibana allows you to generate a URL or embed the dashboard in other platforms.

Step 9: Set Time Ranges and Filters

Dashboards are interactive, and you can:

 Set a time range filter (e.g., last 24 hours, last 7 days).

 Apply additional filters to narrow down the data on the dashboard.

Step 10: Monitor and Update

 Once your dashboard is live, you can continue to monitor the data, make adjustments
to visualizations, and refresh it as needed.
  Kibana dashboards are dynamic and will automatically update if the data in
Elasticsearch is updated in real-time (depending on the data refresh rate).

EXP - 8. Perform the following operations

a. Staging and validating data
b. Building idempotent data pipelines
c. Building atomic data pipelines

a. Staging and Validating Data

1. Staging Data:
o What it is: Staging data is the process of temporarily storing raw data in a
staging area before further processing or transformation. It ensures that the
data is cleaned, pre-processed, and validated before loading it into production
systems.
o How to achieve it:
 Extract data from different sources (databases, APIs, files, etc.).
 Store it in a staging area, typically a raw data lake, cloud storage
(like S3), or a database.
 Format the data: Convert the data into a standardized format for
easier processing (e.g., CSV, Parquet, JSON).
 Data Profiling: Analyze the structure, quality, and completeness of the
raw data to understand its characteristics.
2. Validating Data:
o What it is: Data validation ensures that the data in your pipeline is accurate,
consistent, and meets predefined rules or constraints.
o How to achieve it:
 Schema Validation: Ensure the data matches a predefined schema
(e.g., data types, nullability, range constraints).
 Data Quality Checks: Check for missing values, outliers, duplicates,
and incorrect formats.
 Business Rule Validation: Ensure data satisfies business-specific
rules or logic (e.g., transaction dates must not be in the future).
 Automated Testing: Implement automated data quality tests using
tools like Great Expectations or dbt to ensure ongoing validation.
 Error Handling: If data doesn’t meet validation rules, trigger an alert,
reject the data, or log issues for manual review.

b. Building Idempotent Data Pipelines

1. What it is: Idempotent data pipelines ensure that running the same pipeline multiple
times with the same input does not result in incorrect or inconsistent data. In other
 words, the outcome of the pipeline will remain the same, no matter how many times
it's executed with the same data.
2. How to achieve it:
o Use of Unique Identifiers: Ensure that each record processed in the pipeline
has a unique identifier (ID) so that any duplicated data can be handled
appropriately (e.g., overwriting or ignoring duplicates).
o Upserts (Insert or Update): When inserting data into a database, use upsert
operations that insert new records or update existing ones based on unique
keys. This prevents duplicate records.
o Transactional Integrity: Use database transactions to ensure that operations
like inserts or updates are atomic and roll back in case of failure.
o Avoiding State Duplication: Ensure that intermediate states or outputs from
previous pipeline runs are not reused inappropriately. Use techniques like
time-based partitions or batch IDs to differentiate datasets from different
pipeline executions.
o Idempotent Functions: Make sure that the transformations or calculations
within the pipeline can be repeated without changing the final result, even
when run multiple times.

c. Building Atomic Data Pipelines

1. What it is: Atomic data pipelines refer to pipelines that ensure atomicity, meaning
each pipeline operation is executed completely or not at all. This is crucial for
maintaining the integrity of data processing. If any step of the pipeline fails, all
changes must be rolled back, and no partial data should be left behind.
2. How to achieve it:
o Transactional Integrity: Use database transactions or systems that support
atomic operations (e.g., relational databases or some NoSQL databases) to
group steps together. If one step fails, the transaction is rolled back, ensuring
no partial changes are committed.
o Error Handling and Recovery: Implement robust error handling to capture
failures at any step of the pipeline. If something goes wrong, the pipeline
should be able to either retry or rollback to the last known good state.
o Checkpoints: In long-running or complex pipelines, use checkpoints to
capture the state of the pipeline at various stages. If a failure occurs, you can
restart from the last checkpoint instead of starting over.
o Atomic File Operations: In file-based systems (like HDFS or S3), ensure that
files are fully written or updated atomically. For instance, by writing to a
temporary file and renaming it to the target file only after the process has
completed successfully.
o Minimal External Dependencies: Try to minimize the reliance on external
systems or services that may introduce failure points. If external services are
required, ensure they are idempotent and resilient.

EXP - 9. Version Control with the NiFi Registry

a. Installing and configuring the NiFi Registry

b. Using the Registry in NiFi

c. Versioning your data pipelines

d. Using git-persistence with the NiFi Registry

Here's an overview of the steps to install and configure NiFi Registry, use it within NiFi,
version your data pipelines, and set up git-persistence with NiFi Registry:

a. Installing and Configuring NiFi Registry

1. Download NiFi Registry:

o Visit the Apache NiFi Registry download page and select the appropriate
version for your system.
2. Extract NiFi Registry:
o Once downloaded, extract the NiFi Registry archive:

bash
CopyEdit
tar -xvf nifi-registry-<version>-bin.tar.gz
cd nifi-registry-<version>

3. Configuration:
o Open the conf/nifi-registry.properties file and configure the following
essential settings:
 Web HTTP Port: Set the port for accessing the NiFi Registry UI. The
default is 18080.

properties
CopyEdit
nifi.registry.web.http.port=18080

 Persistent storage location: Set where versioned flows will be stored.

properties
CopyEdit
nifi.registry.repo.directory=/path/to/storage

4. Start NiFi Registry:

o To start the NiFi Registry, run the following command:

bash
CopyEdit
./bin/nifi-registry.sh start

5. Access NiFi Registry UI:

o Open a browser and go to http://localhost:18080 (or the configured URL)
to access the NiFi Registry UI.

b. Using the Registry in NiFi

1. Configure NiFi to connect to the NiFi Registry:

o Open the NiFi instance's conf/nifi.properties file and configure the NiFi
Registry URL:
 properties
CopyEdit
nifi.registry.client.enabled=true
nifi.registry.url=http://localhost:18080

2. Linking NiFi to the Registry:

o In the NiFi UI, go to the "Operate" palette and click on the NiFi Registry tab.
Add the NiFi Registry connection with the configured URL.
3. Accessing Versioned Flows:
o You can now version your NiFi flows directly in NiFi:
 Go to the Flow Configuration of your NiFi processor or flow and link
it to a versioned flow from the NiFi Registry.
4. Registering Flows:
o Once a flow is designed, you can register it with NiFi Registry by clicking the
Version Control option for the flow in NiFi UI.

c. Versioning Your Data Pipelines

1. Create a Versioned Flow:

o After configuring your NiFi Registry, create a Versioned Flow by selecting
the flow and clicking on the Version Control option. You can now create a
new version for that flow.
2. Commit Flow Versions:
o Once your flow is registered, you can commit any changes you make by
selecting the Commit option. This ensures you have a version history and can
track changes.
3. View Versions:
o To view the version history of your flows, navigate to the NiFi Registry UI,
and you will see a list of versioned flows with details about each version.
4. Rollback to Previous Versions:
o If needed, you can revert to a previous version of a flow by selecting the
desired version and clicking on the Rollback option. This ensures the flow is
reverted to its earlier state.

d. Using Git-Persistence with the NiFi Registry

1. Enable Git-Based Persistence:

o NiFi Registry supports storing flow versions in a Git repository. To configure
this, open the conf/nifi-registry.properties file and configure the Git
repository settings:

properties
CopyEdit
nifi.registry.persistence.provider=git
nifi.registry.persistence.git.repo.directory=/path/to/git/
repository
nifi.registry.persistence.git.repo.branch=main

2. Configure Git Credentials:

o If your Git repository is private, you'll need to provide the necessary
credentials. You can do this by configuring the Git credentials in conf/nifi-
 registry.properties or using the standard Git credential storage
mechanisms.
3. Push Flow Versions to Git:
o With Git persistence enabled, every time a version of a flow is committed in
the NiFi Registry, it will be saved to the Git repository, allowing for version
control and collaboration.
4. Cloning and Fetching from Git:
o You can also clone or fetch versions of the NiFi flow directly from the Git
repository by configuring the NiFi Registry to point to the appropriate
repository.

10. Monitoring Data Pipelines

a. Monitoring NiFi in the GUI
b. Monitoring NiFi using processors
c. Monitoring NiFi with Python and the REST API

onitoring NiFi data pipelines is crucial to ensure smooth data flow and troubleshoot any
issues that arise in real-time. Below are different ways you can monitor NiFi pipelines:

EXP - 10. Monitoring Data Pipelines

a. Monitoring NiFi in the GUI

The NiFi GUI provides an intuitive and interactive way to monitor and manage your data
flows. Some key aspects of monitoring in the GUI include:

1. Data Flow Overview:

o NiFi's main canvas shows the flow of data through processors, connections, and
other components.
o Data Provenance: You can track the lifecycle of each data object, seeing exactly
where it’s been, when it arrived, and how it was processed.
o Process Group Monitoring: You can view the health and performance of individual
process groups in your data pipeline.

2. Processor and Connection Status:

o Processor status: Indicates whether a processor is active, idle, or has failed. It
provides indicators such as:
 Running (Green)
 Stopped (Gray)
 Failed (Red)
o Connection status: View the queue sizes in the connections between processors,
showing how many flow files are waiting to be processed.

3. Processor Metrics:
o Monitor various metrics such as:
 Flow File Count
 Bytes In and Out
 Processing Time per processor
o Click on individual processors to get more detailed statistics like success/failure
rates, the number of flow files processed, and error messages.
 4. Bulletins and Alerts:
o NiFi shows bulletins at the top of the screen for any warnings or errors in your data
flow.
o You can filter the bulletins by severity to focus on the most critical issues.

5. NiFi Registry:
o If you are using NiFi Registry, you can monitor versioned flows and ensure that the
changes to data flows are properly tracked.

b. Monitoring NiFi using Processors

You can leverage NiFi processors themselves for monitoring data flows:

1. Monitoring Processor Performance:

o ExecuteScript Processor: You can write custom scripts to collect
performance metrics or log error conditions within the flow. For example,
using Python or Groovy to measure the number of flow files processed or the
processing duration.
o LogAttribute Processor: It logs details about flow files at various stages,
allowing you to track how many files pass through a given point in your flow
and log important information.
2. Utilizing Site-to-Site Processors:
o Site-to-Site allows NiFi instances to communicate with each other. You can monitor
how many flow files are being transferred between instances and check for
bottlenecks.

3. Custom Metrics:
o NiFi provides custom metrics for processors like the QueryDatabaseTable processor
or any custom processors that you might create.
o You can also use ExecuteSQL or ExecuteScript to query your NiFi instance for
custom metrics or flow data.

4. Monitor Queue Sizes:

o Connection Metrics: NiFi provides the ability to monitor how many flow files are in a
queue. You can set up alerting or monitoring actions if a queue size exceeds a
threshold (i.e., the queue is filling up faster than it’s being processed).
o Use a Data Provenance processor to gather lineage and view any bottlenecks.

c. Monitoring NiFi with Python and the REST API

NiFi’s REST API provides extensive capabilities to monitor and manage NiFi
programmatically. You can use Python to automate monitoring tasks. Here's how to integrate
Python with NiFi’s REST API:

1. Getting Processor Status:

o You can query NiFi’s REST API to get the status of processors, process groups, or
connections. For example, use GET /nifi-api/controller/processors to get
information about processor status.

import requests
 base_url = 'http://localhost:8080/nifi-api'
response = requests.get(f'{base_url}/controller/processors')
if response.status_code == 200:
processors = response.json()
for processor in processors['processors']:
print(processor['id'], processor['status'])

2. Processor Metrics:
o For example, if you want to monitor flow file counts or processing times:

processor_id = 'your-processor-id'
response =
requests.get(f'{base_url}/controller/processors/{processor_id}/status
')
status = response.json()
print(f"Processor {processor_id} status: {status['status']}")

3. Querying Processor Bulletins:

o NiFi’s REST API allows you to query for any bulletins related to processor errors or
warnings, providing you with real-time alerts.

response = requests.get(f'{base_url}/controller/bulletins')
bulletins = response.json()
for bulletin in bulletins:
print(f"Bulletin: {bulletin['message']}")

4. Monitoring Process Group Status:

o You can monitor entire process groups to see the flow of data and identify
bottlenecks.

python
CopyEdit
process_group_id = 'your-process-group-id'
response = requests.get(f'{base_url}/flow/process-
groups/{process_group_id}/status')
status = response.json()
print(f"Process Group {process_group_id} status: {status['status']}")

5. Automating Alerts:
o You can create automated monitoring scripts in Python that check processor
statuses periodically and send email alerts or other notifications if certain conditions
are met (e.g., high queue size, failure rates).

6. Tracking Data Provenance:

o NiFi’s API also allows you to fetch data provenance events to track the flow of data
through your system, which can be useful for debugging or auditing purposes.

response = requests.get(f'{base_url}/nifi-api/provenance')

provenance_data = response.json()

print(provenance_data)
EXP- 11. Deploying Data Pipelines

Deploying a data pipeline to production involves a series of steps to ensure that your pipeline
works reliably and efficiently at scale. This includes finalizing the pipeline design, using
configurations like the NiFi variable registry, and finally deploying the pipeline for
continuous operation.

a. Finalizing Your Data Pipelines for Production

When you're ready to move your data pipeline into production, it’s important to finalize
various elements of the pipeline for optimal performance, reliability, and monitoring. The
following tasks should be addressed:

1. Code Review and Testing:

o Ensure that the pipeline logic is thoroughly reviewed and tested.
o Conduct unit tests, integration tests, and performance tests to verify that the
pipeline behaves as expected under load.

2. Error Handling:
o Implement comprehensive error handling to capture and deal with unexpected
failures in the pipeline.
o Set up alerts and logging mechanisms to help monitor the pipeline during
production.

3. Scalability:
o Check whether your pipeline can scale horizontally (across multiple nodes) or
vertically (by adding resources to existing nodes) based on traffic.
o Ensure that resources like memory, CPU, and disk space are appropriately allocated
to handle peak loads.

4. Data Quality:
o Implement checks to monitor data quality, ensuring the integrity and accuracy of the
data being processed.

5. Documentation:
o Document the pipeline architecture, including how each component of the pipeline
functions and interacts.
o Provide clear instructions for troubleshooting and support.

6. Security Considerations:
o Implement security measures such as encryption, secure data transfers, and access
controls to protect sensitive data.
o Ensure compliance with relevant data protection regulations (e.g., GDPR).

7. Version Control and Rollback Strategy:

o Ensure that the pipeline is version-controlled, allowing for easy rollback to a
previous version if something goes wrong after deployment.
b. Using the NiFi Variable Registry

Apache NiFi is a powerful data integration tool that simplifies data pipeline creation and
deployment. To make your NiFi-based data pipelines more adaptable and configurable, you
can use the NiFi Variable Registry.

The NiFi Variable Registry allows you to store and manage variables that can be referenced
in different parts of your NiFi flow. These variables can hold values that you want to reuse
across processors, such as API endpoints, database credentials, or other environment-specific
parameters.

Steps to use NiFi Variable Registry:

1. Accessing the Variable Registry:

o Navigate to the NiFi UI, and from the top-right menu, select "Variable Registry".
o Here, you can create variables to be used globally within your NiFi data flow.

2. Creating Variables:
o Click "Add Variable" to create a new variable.
o Provide a variable name and its value (e.g., API_KEY=xyz123).
o You can create variables for things like database credentials, server URLs, or API
keys.

3. Referencing Variables in Processors:

o In your NiFi processors (e.g., ExecuteScript, PutSQL, etc.), you can use the
variable by referencing it in the form of ${variable_name}.
o For example, you can use ${API_KEY} in the Processor Properties to dynamically
reference the API key set in the Variable Registry.

4. Environment-specific Variables:
o You can also define environment-specific variables by creating separate variable
registries for each environment (e.g., dev, staging, production).

5. Version Control and Deployment:

o Variables can be defined across different environments, and changes to the
variables can be made without altering the core flow of the NiFi pipeline.

By using the Variable Registry, you decouple configuration from your NiFi flow, making it
easier to modify settings for different environments (such as production or development)
without modifying the actual flow design.

c. Deploying Your Data Pipelines

Once your data pipeline is ready, the final step is deployment. Deploying a data pipeline
means making it available for continuous operation. The deployment process can differ
depending on the environment and tools used (e.g., NiFi, Kubernetes, or cloud services). The
following is a general approach to deploying data pipelines:
1. Containerization (Optional):
o If using a platform like Kubernetes or Docker, package your data pipeline into
containers. This ensures portability and ease of deployment.
o Use tools like Dockerfile to containerize your NiFi or other pipeline components.

2. Deploy to the Target Environment:

o NiFi Deployment: If you’re using NiFi, deploy the flow to a NiFi instance running in
your production environment.
o Cloud Deployment: For cloud environments (e.g., AWS, Google Cloud, Azure), you
can deploy the pipeline using cloud-native tools (e.g., AWS Lambda, Google Cloud
Functions, or Databricks) or container orchestration services like Kubernetes.
o CI/CD Pipelines: Set up a Continuous Integration/Continuous Deployment (CI/CD)
pipeline to automatically deploy updates and monitor the pipeline in production.

3. Configuration Management:
o Leverage configuration management tools (e.g., Ansible, Terraform, or Chef) to
deploy and manage pipeline configurations across different environments.
o Set environment-specific variables to ensure that the pipeline is properly configured
for production.

4. Monitoring and Logging:

o Use tools like Prometheus, Grafana, or NiFi’s built-in monitoring to keep an eye on
your data pipeline’s health and performance.
o Set up proper logging mechanisms for auditing and troubleshooting. NiFi has built-in
logging features, but you can also integrate it with external log management
systems.

5. Scaling and Load Balancing:

o For large-scale production deployments, you may need to set up horizontal scaling,
load balancing, and failover mechanisms. In cloud environments, auto-scaling groups
or Kubernetes can automatically scale your pipeline as demand fluctuates.
o Ensure that your pipeline can handle failover scenarios by using multiple processing
nodes or fault-tolerant components.

6. Testing in Production (Canary Releases, Blue-Green Deployment):

o Use canary releases to test the pipeline with a small subset of traffic before rolling it
out to the full production environment.
o Consider a blue-green deployment strategy, where you maintain two environments
(one "blue" for production and one "green" for testing) and switch traffic between
them to minimize downtime during deployment.

7. Post-deployment Verification:
o After deployment, verify that the pipeline is functioning correctly by checking
output, logs, and performance metrics.
o Monitor system resources and check for errors or unexpected behaviors.

8. Maintainability and Updates:

o Plan for regular updates to the pipeline. Regularly update processing logic, patches,
or configuration changes, and use versioning to avoid breaking changes.
o Monitor the system continuously to detect performance degradation and plan for
optimizations.
EXP - 12. Building a Production Data Pipeline
a. Creating a test and production environment
b. Building a production data pipeline
c. Deploying a data pipeline in production

uilding a production data pipeline involves several key steps to ensure data flows smoothly
from its source to its final destination, while also ensuring scalability, reliability, and proper
monitoring. Here's an overview of each step involved in creating a production-grade data
pipeline:

a. Creating a Test and Production Environment

Before diving into building a pipeline, it's essential to have both test and production
environments, as they help ensure that your pipeline is functioning properly and can handle
the scale of production workloads.

1. Test Environment:
o Isolated: The test environment should be isolated from the production
environment to avoid disruptions.
o Mimic Production: It should replicate the production environment as closely
as possible, in terms of both data volume and structure, but with anonymized
or smaller-scale data.
o Infrastructure as Code (IaC): Use tools like Terraform, CloudFormation, or
Docker Compose to set up repeatable and version-controlled infrastructure for
the test environment.
o Automated Testing: Run unit tests, integration tests, and data validation
checks on data pipelines to ensure they function as expected before
deployment to production.
2. Production Environment:
o Scalable Infrastructure: Set up cloud-based or on-premise infrastructure to
handle production-scale data loads. Utilize elastic services (e.g., AWS, GCP,
Azure) to scale resources dynamically as demand increases.
o Data Storage: Use robust storage solutions like Amazon S3, Google Cloud
Storage, or HDFS, depending on your needs for durability and access speed.
o Monitoring & Logging: Implement comprehensive monitoring tools (e.g.,
Prometheus, Grafana, DataDog) and logging (e.g., ELK stack, Fluentd) to
track performance, errors, and issues in real-time.
o Security: Ensure data encryption, access control, and data masking, especially
if dealing with sensitive information.

b. Building a Production Data Pipeline

1. Data Ingestion:
o Collect raw data from various sources like databases, APIs, files, or streaming
services (e.g., Kafka, AWS Kinesis, Google Pub/Sub).
o Use batch processing (e.g., Apache Spark, AWS Glue) or real-time streaming
(e.g., Apache Flink, Kafka Streams) depending on your use case.
2. Data Transformation:
 o Data cleaning, enrichment, and transformation happen here. Tools like
Apache Airflow, dbt (Data Build Tool), or AWS Glue are commonly used
for orchestration and transformation.
o ETL vs ELT: Decide whether you'll perform ETL (Extract, Transform, Load)
or ELT (Extract, Load, Transform) based on the tools and infrastructure
available.
o Ensure data transformations are idempotent (i.e., rerun without causing
inconsistencies) to handle retries in case of failure.
3. Data Storage:
o Store transformed data in databases or data lakes (e.g., Amazon Redshift,
Google BigQuery, Snowflake) or NoSQL solutions (e.g., MongoDB,
Cassandra).
o Consider partitioning, indexing, and using columnar storage formats (e.g.,
Parquet, ORC) for optimized query performance.
4. Data Quality and Validation:
o Implement data validation checks to ensure data integrity, such as schema
validation and range checks.
o Tools like Great Expectations and Deequ can be used for automated data
validation.
5. Data Orchestration:
o Orchestrate tasks in the pipeline using tools like Apache Airflow, Dagster, or
Kubeflow Pipelines to schedule and automate workflows.
o Ensure the pipeline can handle failure and recovery, so tasks can retry in case
of issues.

c. Deploying a Data Pipeline in Production

1. Automation and CI/CD for Data Pipelines:

o Implement Continuous Integration (CI) and Continuous Deployment (CD)
pipelines for automated testing and deployment. Tools like GitLab CI,
CircleCI, or Jenkins can help automate deployment to production.
o Ensure code changes are versioned, and use pull requests and code reviews to
maintain quality.
2. Monitoring and Alerts:
o Set up robust monitoring for the data pipeline to ensure it runs smoothly in
production. Use tools like Prometheus, Grafana, and Cloud-native
monitoring tools for visibility.
o Create alerts for failures, performance bottlenecks, or unusual data patterns.
For example, you can set up alerts if certain thresholds are breached (e.g., data
delay, transformation errors).
3. Data Lineage and Auditing:
o Implement data lineage tracking (e.g., using tools like Amundsen,
OpenLineage, or DataHub) to trace the flow of data across the pipeline and
understand how data transforms and is consumed.
o Audit logs help track the entire process, providing insights into data
transformations and access history.
4. Scaling the Pipeline:
o Ensure the pipeline is scalable. Use horizontal scaling to add more workers or
resources, or use cloud-native auto-scaling features.
 o Partition large datasets and ensure efficient data storage practices for both
batch and real-time processing workloads.
5. Backup and Disaster Recovery:
o Implement backup strategies for critical data and ensure that data pipelines are
fault-tolerant.
o Set up disaster recovery plans to quickly recover the pipeline in case of failure.
6. Security and Compliance:
o Apply necessary access controls using Identity and Access Management
(IAM) policies.
o Ensure the pipeline adheres to industry standards and regulatory requirements
(e.g., GDPR, HIPAA) for data handling, storage, and transmission.