DEV Community: Angélica Beatriz ROMERO ROQUE

Building a Real-Time Data Pipeline App with Change Data Capture Tools: Debezium, Kafka, and NiFi

Angélica Beatriz ROMERO ROQUE — Tue, 17 Dec 2024 22:50:53 +0000

Change Data Capture (CDC) has become a critical technique for modern data integration, allowing organizations to track and propagate data changes across different systems in real-time. In this article, we'll explore how to build a comprehensive CDC solution using powerful open-source tools like Debezium, Apache Kafka, and Apache NiFi

Key Technologies in Our CDC Stack

Debezium: An open-source platform for change data capture that supports multiple database sources.
Apache Kafka: A distributed streaming platform that serves as the central nervous system for our data pipeline.
Apache NiFi: A data flow management tool that helps us route, transform, and process data streams.

Architecture Overview
Our proposed architecture follows these key steps:

Capture database changes using Debezium
Stream changes through Kafka
Process and route data using NiFi
Store or further process the transformed data

Sample Implementation Approach

from confluent_kafka import Consumer, Producer
import json
import debezium

class CDCDataPipeline:
    def __init__(self, source_db, kafka_bootstrap_servers):
        """
        Initialize CDC pipeline with database source and Kafka configuration

        :param source_db: Source database connection details
        :param kafka_bootstrap_servers: Kafka broker addresses
        """
        self.source_db = source_db
        self.kafka_servers = kafka_bootstrap_servers

        # Debezium connector configuration
        self.debezium_config = {
            'connector.class': 'io.debezium.connector.mysql.MySqlConnector',
            'tasks.max': '1',
            'database.hostname': source_db['host'],
            'database.port': source_db['port'],
            'database.user': source_db['username'],
            'database.password': source_db['password'],
            'database.server.name': 'my-source-database',
            'database.include.list': source_db['database']
        }

    def start_capture(self):
        """
        Start change data capture process
        """
        # Configure Kafka producer for streaming changes
        producer = Producer({
            'bootstrap.servers': self.kafka_servers,
            'client.id': 'cdc-change-producer'
        })

        # Set up Debezium connector
        def handle_record(record):
            """
            Process each captured change record
            """
            # Transform record and publish to Kafka
            change_event = {
                'source': record.source(),
                'operation': record.operation(),
                'data': record.after()
            }

            producer.produce(
                topic='database-changes', 
                value=json.dumps(change_event)
            )

        # Start Debezium connector
        debezium.start_connector(
            config=self.debezium_config,
            record_handler=handle_record
        )

# Example usage
source_database = {
    'host': 'localhost',
    'port': 3306,
    'username': 'cdc_user',
    'password': 'secure_password',
    'database': 'customer_db'
}

pipeline = CDCDataPipeline(
    source_database, 
    kafka_bootstrap_servers='localhost:9092'
)
pipeline.start_capture()

Detailed Implementation Steps

Database Source Configuration The first step involves configuring Debezium to connect to your source database. This requires:

Proper database user permissions
Network connectivity
Enabling binary logging (for MySQL)

Kafka as a Streaming Platform Apache Kafka acts as a central message broker, capturing and storing change events. Key considerations include:

Configuring topic partitions
Setting up appropriate retention policies
Implementing exactly-once processing semantics

Data Transformation with NiFi Apache NiFi provides powerful data routing and transformation capabilities:

Filter and route change events
Apply data enrichment
Handle complex transformation logic

Challenges and Best Practices

Handling Schema Changes: Implement robust schema evolution strategies
Performance Optimization: Use appropriate partitioning and compression
Error Handling: Implement comprehensive error tracking and retry mechanisms

GitHub Repository

I've created a sample implementation that you can explore and use as a reference. The complete code and additional documentation can be found at:
GitHub Repository: https://github.com/Angelica-R/cdc-data-pipeline

Conclusion
Building a Change Data Capture solution requires careful architectural design and selection of appropriate tools. By leveraging Debezium, Kafka, and NiFi, you can create a robust, scalable data integration platform that provides real-time insights into your data changes.

Building an App with a Cloud NoSQL Database (Avoiding DynamoDB or Cosmos DB)

Angélica Beatriz ROMERO ROQUE — Tue, 17 Dec 2024 22:37:37 +0000

Cloud-based NoSQL databases have become increasingly popular due to their scalability, flexibility, and ability to handle unstructured data. However, some developers may need to avoid commonly used services like AWS DynamoDB or Azure Cosmos DB, either for cost, compatibility, or preference reasons. In this article, we'll explore how to build an app using an alternative cloud NoSQL database, Firebase Firestore, and demonstrate its implementation with sample code.

Why Use a Cloud NoSQL Database?
NoSQL databases are ideal for:

Scalability: Easily handle large volumes of data across distributed systems.
Flexibility: Store data in various formats (JSON, key-value, document, etc.).
Real-Time Sync: Many NoSQL databases, like Firestore, provide real-time capabilities.
Ease of Use: Simplified data modeling compared to relational databases. In this article, we’ll use Firebase Firestore, a document-based database that offers real-time synchronization, scalability, and serverless management, making it an excellent alternative to DynamoDB or Cosmos DB.

Technologies We'll Use

Database: Firebase Firestore
Frontend: React.js
Backend: Node.js (optional)
Cloud Environment: Firebase Hosting (optional)

Setting Up Firebase Firestore
Step 1: Create a Firebase Project

Visit Firebase Console.
Create a new project and enable Firestore under the "Build" section.

Step 2: Install Firebase SDK
Add Firebase to your project by installing the Firebase SDK in your React app:
npm install firebase

Step 3: Configure Firebase in Your App
Create a firebaseConfig.js file to initialize Firebase:
`// firebaseConfig.js
import { initializeApp } from 'firebase/app';
import { getFirestore } from 'firebase/firestore';

const firebaseConfig = {
apiKey: "your-api-key",
authDomain: "your-app.firebaseapp.com",
projectId: "your-project-id",
storageBucket: "your-project-id.appspot.com",
messagingSenderId: "your-sender-id",
appId: "your-app-id"
};

// Initialize Firebase
const app = initializeApp(firebaseConfig);
const db = getFirestore(app);

export default db;`

Building the App
Project Structure

cloud-no-sql-app/
│-- src/
│   ├── components/
│   │   ├── AddItem.js
│   │   ├── ItemList.js
│   └── firebaseConfig.js
│-- App.js
└── package.json

Backend: Firestore CRUD Operations
Adding Data
To add a document to a Firestore collection:

import { collection, addDoc } from 'firebase/firestore';
import db from './firebaseConfig';

const addItem = async (item) => {
    try {
        const docRef = await addDoc(collection(db, 'items'), item);
        console.log('Document written with ID:', docRef.id);
    } catch (e) {
        console.error('Error adding document:', e);
    }
};

Reading Data
To fetch all documents from a Firestore collection:

import { collection, getDocs } from 'firebase/firestore';

const fetchItems = async () => {
    const querySnapshot = await getDocs(collection(db, 'items'));
    querySnapshot.forEach((doc) => {
        console.log(`${doc.id} =>`, doc.data());
    });
};

Deleting Data
To delete a document:

import { doc, deleteDoc } from 'firebase/firestore';

const deleteItem = async (id) => {
    await deleteDoc(doc(db, 'items', id));
    console.log('Document deleted with ID:', id);
};

Frontend: React Components
AddItem Component

import React, { useState } from 'react';
import { addItem } from './firebaseConfig';

const AddItem = () => {
    const [item, setItem] = useState('');

    const handleSubmit = async (e) => {
        e.preventDefault();
        await addItem({ name: item });
        setItem('');
    };

    return (
        <form onSubmit={handleSubmit}>
            <input
                type="text"
                value={item}
                onChange={(e) => setItem(e.target.value)}
                placeholder="Enter item name"
            />
            <button type="submit">Add Item</button>
        </form>
    );
};

export default AddItem;

ItemList Component

import React, { useState, useEffect } from 'react';
import { fetchItems } from './firebaseConfig';

const ItemList = () => {
    const [items, setItems] = useState([]);

    useEffect(() => {
        const getItems = async () => {
            const data = await fetchItems();
            setItems(data);
        };
        getItems();
    }, []);

    return (
        <ul>
            {items.map((item) => (
                <li key={item.id}>{item.name}</li>
            ))}
        </ul>
    );
};

export default ItemList;

Running the App

Start your development server:

npm start

Add items and view them in real time via the Firestore Console or your app.

GitHub Repository
You can find the complete source code for this project on GitHub:
GitHub Repository: https://github.com/Angelica-R/cloud-no-sql-app.git

Conclusion
Using Firebase Firestore, you can easily build scalable applications without relying on SQL Server, DynamoDB, or Cosmos DB. Firestore's real-time synchronization and simplicity make it a strong choice for projects that require cloud-hosted NoSQL databases. By following this tutorial, you’ve learned how to set up Firestore, perform CRUD operations, and integrate it into a React app.

Build an Application Without SQL Server Database (Avoiding RPrometheusedis, MongoDB, and )

Angélica Beatriz ROMERO ROQUE — Tue, 17 Dec 2024 22:05:25 +0000

In the world of software development, databases are one of the fundamental pillars for storing and managing information. However, there are cases where applications do not require a conventional database such as SQL Server or NoSQL solutions like Redis, MongoDB, or Prometheus. This can happen when the application is small, does not need persistent data, or when a simplified and lightweight solution is sought.

In this article, we will explore how to build a functional application without using databases. We will use flat files (JSON) as temporary storage, which is useful for small projects or applications with no large scalability requirements.

1. Reasons to Avoid Traditional Databases

Although SQL Server, Redis, and MongoDB are popular choices, they can be unnecessary in the following scenarios:

Small applications or prototypes.
Temporary storage requirements.
Avoiding costs associated with cloud services.
Need for a simple and lightweight implementation.
Lack of infrastructure or access to servers. A common alternative is to use flat files, like JSON, to store data locally on the server.

2. Technologies to Use
For this example, we will build a simple application using:

Language: Node.js (JavaScript)
Storage: Local JSON files
Framework: Express.js to create a simple HTTP server
Development Tools: Visual Studio Code and npm We will not use any database management system or NoSQL solutions.

3. Project Structure
The file structure will be straightforward:
my-app/ │-- server.js │-- data/ │ └── data.json └── package.json

server.js: Main file where the Express server is implemented.
data.json: Local storage file.
package.json: Dependency configuration file.

4. Code Implementation
4.1 Install Dependencies
Initialize a Node.js project and install Express.js:
mkdir my-app cd my-app npm init -y npm install express
4.2 Create the JSON Data File

In the data folder, create the data.json file:
[ { "id": 1, "name": "Juan", "age": 25 }, { "id": 2, "name": "Pedro", "age": 30 } ]

4.3 Implement the Express Server

In server.js, implement a server that reads and writes data to the JSON file:
`const express = require('express');
const fs = require('fs');
const app = express();
const PORT = 3000;

app.use(express.json());

const filePath = './data/data.json';

// Get all data
app.get('/users', (req, res) => {
fs.readFile(filePath, 'utf8', (err, data) => {
if (err) return res.status(500).send('Error reading data');
res.json(JSON.parse(data));
});
});

// Add a new user
app.post('/users', (req, res) => {
const newUser = req.body;

fs.readFile(filePath, 'utf8', (err, data) => {
    if (err) return res.status(500).send('Error reading data');

    const users = JSON.parse(data);
    users.push({ id: users.length + 1, ...newUser });

    fs.writeFile(filePath, JSON.stringify(users, null, 2), (err) => {
        if (err) return res.status(500).send('Error saving data');
        res.status(201).send('User added successfully');
    });
});

});

app.listen(PORT, () => {
console.log(Server running at http://localhost:${PORT});
});`

5. How It Works

Data Reading: The GET /users endpoint reads the JSON file and returns its content as a response.
Data Writing: The POST /users endpoint adds a new user to the JSON file and saves the changes.

6. Testing the Application

Start the server: node server.js
Use Postman or your browser to test the endpoints:
GET http://localhost:3000/users returns the list of users.
POST http://localhost:3000/users with a JSON body like: { "name": "Luis", "age": 28 } adds a new user. 7. Source Code on GitHub You can find the complete code in the following GitHub repository: GitHub Repository: https://github.com/Angelica-R/App-without-SQL-Server-Database.git

8. Conclusion
Building an application without databases like SQL Server, Redis, or MongoDB is a valid option for small and simple applications. Using JSON files provides a lightweight and easy-to-implement solution, allowing data management in a temporary or local form. This technique is ideal for prototypes or environments where a complex database is not needed.

Furthermore, this approach allows for future scalability, as it is always possible to migrate to a traditional database when the project requires it.