DEV Community: JhonnyARM

Building an App with Change Data Capture (CDC) Tool

JhonnyARM — Thu, 12 Dec 2024 16:03:40 +0000

Introduction

In today's modern applications, the need to track changes in real-time data is crucial for a variety of use cases, including analytics, real-time processing, and system integration. One of the most efficient ways to achieve this is by implementing Change Data Capture (CDC).

This article explores how to build an application that leverages a CDC tool, such as Debezium, to track changes in a database, capture them in real-time, and forward those changes to a messaging system like Apache Kafka for processing. We'll walk through the process and architecture of using CDC to build an event-driven app.

What is Change Data Capture (CDC)?

Change Data Capture (CDC) is a technique used to identify and capture changes made to a database, such as INSERT, UPDATE, and DELETE operations. CDC allows applications to react to changes in the data layer in real-time without the need for constant polling or complex queries.

Debezium is one of the most popular open-source CDC tools. It captures changes from databases like MySQL, PostgreSQL, and others, and streams those changes into Apache Kafka topics.

Components of the Architecture

Before diving into the implementation, let's break down the architecture components:

Database:

The data source where your app stores information. Common databases used in this architecture include MySQL, PostgreSQL, etc.
Changes to the data (like INSERT, UPDATE, DELETE) are captured by the CDC tool.

CDC Tool (Debezium):

Debezium listens to the database's changes (using the binlog for MySQL, for example) and captures them as events.
It then publishes those events to an Apache Kafka topic, making them available for other services to consume.

Kafka:

Apache Kafka acts as the message broker that stores and streams these data changes in topics.
Kafka ensures that data changes are available to multiple consumers without creating tight coupling between your database and consumers.

Consumer (App or API):

A consumer application subscribes to the Kafka topic and processes the data.
The consumer can perform tasks like transforming the data, storing it in a different database (NoSQL), triggering notifications, or performing additional business logic.

NoSQL Database:

In some use cases, the consumer might store the processed changes in a NoSQL database like MongoDB or Elasticsearch, where the data can be queried and analyzed efficiently.

Architecture

Here's an overview of the components and how they interact:

Step-by-Step Process

Set up your Database:
- For this example, let's assume we are using MySQL. Set up a MySQL database where your application stores the data.
Configure Debezium:
- Install and configure Debezium to monitor the MySQL database. You'll need to configure the binlog in MySQL and ensure that Debezium is connected to your Kafka instance.
Kafka Configuration:
- Set up Kafka to handle the messages that Debezium produces. Create topics where the CDC events will be published.
Create Consumers:

Develop a Kafka consumer that listens to the Kafka topics where Debezium sends the events. This consumer processes the data (e.g., transforming the data or storing it in a NoSQL database).

Persist to NoSQL:
- If needed, the consumer stores the processed data in a NoSQL database (like MongoDB or Elasticsearch) for further analysis or querying.

Results

Github:
https://github.com/JhonnyARM/ArticuloU3
Docker Compose

Swagger Logs & Products endpoints

Benefits of Using CDC in Your Application

Real-Time Data Processing: CDC enables your app to react to changes as they happen, providing real-time insights and updates.
Scalability: By using Kafka as a message broker, you can scale your architecture horizontally, adding more consumers as needed.
Decoupling: The CDC tool and Kafka decouple your database from the rest of your application. This reduces the risk of bottlenecks and allows for more efficient data handling.
Flexibility: Storing the data in a NoSQL database after processing allows for more flexible querying and faster access to the data.

Conclusion

Building an app with a Change Data Capture (CDC) tool like Debezium enables your application to react to database changes in real-time. By integrating Debezium with Apache Kafka and NoSQL databases, you can implement an efficient, scalable, and flexible architecture that handles large volumes of data changes with minimal latency.

Building an App with a Cloud NoSQL Database

JhonnyARM — Tue, 10 Dec 2024 21:36:09 +0000

Firebase

In this article, we will create an application using Firebase, a cloud-based platform that provides a NoSQL database and other tools to simplify backend development.

What is Firebase?

Firebase provides developers with tools and services to build and manage applications efficiently. Key Firebase services include:

Firestore: A NoSQL document-based database. It stores data in structures called collections, which contain documents. Each document is a flexible container that holds data as key-value pairs. This design is ideal for applications requiring real-time synchronization and scalability.
Authentication: User authentication via email, social logins, or custom methods.
Cloud Functions: Serverless backend logic that runs in the cloud.
Cloud Storage: A powerful file storage solution.

Setting Up Firebase

Before integrating Firebase into your project, set up Firebase by following these steps:

1. Create a Firebase Project

Go to the Firebase Console.
Click Create Project, provide a project name, and follow the setup wizard.

2. Enable Firestore

Navigate to Firestore Database in the Firebase Console.
Click Create Database, and select:
- Test Mode: Suitable for development.
- Production Mode: Requires secure rules.

3. Generate Service Account Key

Go to Project Settings > Service Accounts.
Click Generate New Private Key.
Save the generated JSON file (e.g., Config/serviceAccountKey.json) to your project folder.

Integrating Firebase Firestore with .NET Core

Step 1: Install Required NuGet Packages

Install the necessary Firebase and Firestore NuGet packages:

# Firebase Admin SDK
dotnet add package FirebaseAdmin

# Firestore Integration
dotnet add package Google.Cloud.Firestore --version 3.9.0

Step 2: Configure Firebase in .NET Core

Create a service to initialize Firebase and Firestore.

File: Services/FirebaseService.cs

using FirebaseAdmin;
using Google.Apis.Auth.OAuth2;
using Google.Cloud.Firestore;

public class FirebaseService
{
    public FirestoreDb Firestore { get; private set; }

    public FirebaseService()
    {
        string pathToServiceAccountKey = "Config/serviceAccountKey.json";

        if (FirebaseApp.DefaultInstance == null)
        {
            FirebaseApp.Create(new AppOptions()
            {
                Credential = GoogleCredential.FromFile(pathToServiceAccountKey),
            });
        }

        Firestore = FirestoreDb.Create("your-project-id"); // Replace with your Firebase project ID
        Console.WriteLine("Firestore initialized successfully.");
    }
}

Step 3: Create a Controller for Firestore

Create a controller to handle HTTP POST requests for adding data to Firestore. This example ensures the body of the request uses a structured JSON format.

File: Controllers/FirebaseController.cs

using Google.Cloud.Firestore;
using Microsoft.AspNetCore.Mvc;

[ApiController]
[Route("api/[controller]")]
public class FirebaseController : ControllerBase
{
    private readonly FirebaseService _firebaseService;

    public FirebaseController(FirebaseService firebaseService)
    {
        _firebaseService = firebaseService;
    }

    [HttpPost("add")]
    public async Task<IActionResult> AddDocument([FromBody] FirestoreData data)
    {
        try
        {
            if (data == null || string.IsNullOrEmpty(data.Nombre))
            {
                return BadRequest("The 'Nombre' field is required.");
            }

            CollectionReference collection = _firebaseService.Firestore.Collection("new-collection");
            DocumentReference docRef = await collection.AddAsync(data);

            return Ok($"Document added with ID: {docRef.Id}");
        }
        catch (Exception ex)
        {
            return StatusCode(500, $"Error: {ex.Message}");
        }
    }
}

public class FirestoreData
{
    public string Nombre { get; set; }
}

Step 4: Configure Dependency Injection

var builder = WebApplication.CreateBuilder(args);

// Add services to the container
builder.Services.AddSingleton<FirebaseService>();
builder.Services.AddControllers();

var app = builder.Build();

app.MapControllers();
app.Run();

Step 5: Test the POST Endpoint

Use tools like Postman, Swagger, or cURL to test the POST /api/Firebase/add endpoint.

Example Request:

curl -X POST "http://localhost:5000/api/Firebase/add" \
-H "Content-Type: application/json" \
-d '{ "nombre": "Sample Data" }'

Expected Response:

"Document added with ID: abc123"

Conclusion

Integrating Firebase Firestore provides a robust and scalable solution for managing data.

With Firebase, backend development becomes more efficient, enabling you to focus on delivering features rather than managing infrastructure.

Building an APP with NOSQL Database

JhonnyARM — Sat, 26 Oct 2024 03:39:44 +0000

Introduction

DynamoDB

DynamoDB is a fully managed NoSQL database service provided by Amazon Web Services (AWS). It is designed to handle large amounts of data and provides fast and predictable performance with seamless scalability.

Managed Service: DynamoDB is fully managed, meaning that AWS handles operational tasks such as hardware provisioning, setup, configuration, and backups.
NoSQL Database: It is a NoSQL database, allowing for flexible data models and horizontal scaling, making it suitable for applications with varying data structures.
Performance: DynamoDB offers single-digit millisecond response times, ideal for applications that require high-speed data access.
Scalability: The service can automatically scale up and down to accommodate varying workloads, allowing applications to handle large volumes of data and traffic without manual intervention.
Global Tables: DynamoDB supports global tables, enabling multi-region, fully replicated tables that provide low-latency access to data across different geographical locations.
Integration with AWS Services: DynamoDB integrates seamlessly with other AWS services such as Lambda, API Gateway, and S3, allowing for the development of complex applications in the cloud.
Security and Compliance: It provides built-in security features, including encryption at rest and in transit, and supports AWS Identity and Access Management (IAM) for fine-grained access control.

Running Local DynamoDB

We have to go to this URL and download the zip
https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DynamoDBLocal.DownloadingAndRunning.html

After you download the archive, extract the contents and copy the extracted directory to a location of your choice.

To start DynamoDB on your computer, open a command prompt window, navigate to the directory where you extracted DynamoDBLocal.jar, and enter the following command.

java -Djava.library.path=./DynamoDBLocal_lib -jar DynamoDBLocal.jar -sharedDb

Code Example

Now we are going to test a simple web application, using python and boto3 libraries with flask.

Requirements

Python
Boto3, Flask
Docker
DynamoDBLocal files ### Python Code
Project structure

EjemploDynamoDB/
└── app/
│    ├── templates/
│    │   ├── index.html
│    │   └── add_item.html
│    ├── app.py
└── docker-compose.yml
└── Dockerfile

from flask import Flask, render_template, request, redirect, url_for
import boto3
from botocore.exceptions import ClientError

app = Flask(__name__)


dynamodb = boto3.resource(
    'dynamodb',
    endpoint_url="http://dynamodb-local:8000",
    region_name="us-west-2",
    aws_access_key_id="DUMMYIDEXAMPLE",
    aws_secret_access_key="DUMMYEXAMPLEKEY"
)

table_name = "TestTable"


@app.route('/')
def index():
    table = dynamodb.Table(table_name)
    try:
        response = table.scan()
        items = response.get('Items', [])
        return render_template('index.html', items=items)
    except ClientError as e:
        return f"Error: {e}"


@app.route('/add', methods=['GET', 'POST'])
def add_item():
    if request.method == 'POST':
        id = request.form['id']
        nombre = request.form['nombre']
        valor = request.form['valor']
        table = dynamodb.Table(table_name)
        item = {'id': id, 'nombre': nombre, 'valor': valor}
        try:
            table.put_item(Item=item)
            return redirect(url_for('index'))
        except ClientError as e:
            return f"Error: {e}"
    return render_template('add_item.html')

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

index.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Lista de Elementos</title>
</head>
<body>
    <h1>Elementos en DynamoDB</h1>
    <a href="{{ url_for('add_item') }}">Agregar nuevo elemento</a>
    <ul>
        {% for item in items %}
            <li>{{ item['id'] }} - {{ item['nombre'] }}: {{ item['valor'] }}</li>
        {% endfor %}
    </ul>
</body>
</html>

add_item.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Lista de Elementos</title>
</head>
<body>
    <h1>Elementos en DynamoDB</h1>
    <a href="{{ url_for('add_item') }}">Agregar nuevo elemento</a>
    <ul>
        {% for item in items %}
            <li>{{ item['id'] }} - {{ item['nombre'] }}: {{ item['valor'] }}</li>
        {% endfor %}
    </ul>
</body>
</html>

dockerfile


FROM python:3.9-slim


WORKDIR /app


COPY app/app.py /app/app.py
COPY app/templates /app/templates


RUN pip install boto3 flask


CMD ["python", "app.py"]

dockerfile


version: '3.8'
services:
  dynamodb-local:
    image: "amazon/dynamodb-local:latest"
    container_name: dynamodb-local
    command: "-jar DynamoDBLocal.jar -sharedDb -dbPath ./data"
    ports:
      - "8000:8000"
    volumes:
      - "./docker/dynamodb:/home/dynamodblocal/data"
    working_dir: /home/dynamodblocal

  app-python:
    build:
      context: .
      dockerfile: Dockerfile  # Dockerfile en la raíz del proyecto
    container_name: app-python
    depends_on:
      - dynamodb-local
    environment:
      AWS_ACCESS_KEY_ID: 'DUMMYIDEXAMPLE'
      AWS_SECRET_ACCESS_KEY: 'DUMMYEXAMPLEKEY'
    ports:
      - "8080:8080"
    command: >
      sh -c "pip install boto3 flask && python app.py"

Run Docker Compose

In the console run the following command.

docker-compose up -d --build

Results

Build Generative AI Chatbot

JhonnyARM — Tue, 09 Jul 2024 22:15:23 +0000

Introduction

In this article, you will learn how to build a generative AI chatbot that leverages OpenAI's GPT-3.5 to provide personalized and accurate responses to user queries. By integrating a pretrained LLM with a dynamic SQLite database, you'll create an intelligent chatbot capable of handling domain-specific questions. This tutorial will cover setting up the environment, configuring the AI model, and building an interactive user interface using Gradio.

Why Use Generative AI in Chatbots?

Using Generative AI in chatbots offers several key advantages, including:

Personalized Responses:

Generative AI enables chatbots to generate responses that are tailored and relevant to each user's specific queries, leading to a more engaging and satisfying user experience.

Natural Language Understanding:

Generative AI models like GPT-3.5 have advanced natural language understanding capabilities, allowing the chatbot to comprehend complex queries and provide accurate answers.

Contextual Awareness:

These AI models can maintain context across interactions, which helps in providing coherent and contextually appropriate responses even in multi-turn conversations.

Requirements

To successfully build and deploy this generative AI chatbot, you should have a basic understanding or knowledge of the following:

Basic understanding of Large Language Models (LLMs) and how they can be applied to natural language processing tasks.
Basic knowledge of Python programming, including working with libraries and writing scripts.
Familiarity with how to use APIs, including making requests and handling responses.

Tools and Technologies

OpenAI GPT-3.5: Used for interpreting natural language queries and generating SQL queries.
LangChain: Facilitates integration with OpenAI and helps create SQL agents.
SQLAlchemy: Manages database interactions and operations.
Pandas: Handles reading and processing of Excel files.
Gradio: Creates interactive user interfaces.
SQLite: Acts as the database to store and query data.
Dotenv: Loads environment variables from a .env file.

Flow diagram

The following diagram illustrates how this chatbot operates.

Project Setup

Create requirements.txt and copy the following code.

openai==0.27.0
langchain==0.0.152
sqlalchemy==1.4.41
pandas==1.4.3
gradio==2.7.5
python-dotenv==0.20.0

Install Dependencies: Using requirements.txt.

pip install -r requirements.txt

Create the .env File:


OPENAI_API_KEY=OPENAI-TOKEN-HERE
OPENAI_API_BASE=https://api.openai.com/v1

Replace YOUR-OPENAI-API-KEY with your actual API key obtained from OpenAI.

Create a file called "app"

import os
import gradio as gr
from langchain_community.chat_models import ChatOpenAI
from langchain_community.agent_toolkits.sql.base import create_sql_agent
from langchain_community.utilities import SQLDatabase
from langchain_community.tools import Tool
from langchain.memory import ConversationBufferMemory
from dotenv import load_dotenv
from sqlalchemy import create_engine, inspect, text
from loader import create_demand_table

# Load environment variables
load_dotenv()

# Configure the language model with OpenAI directly
llm = ChatOpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="gpt-3.5-turbo-0125",
    temperature=0.1,
)

# Initialize the database engine
engine = create_engine('sqlite:///db.sqlite3', echo=True)

# Function to process the uploaded file and create a table
def process_file(file):
    db = create_demand_table(engine, 'dynamic_table', file.name)
    return db

# Define a function to execute SQL queries
def run_sql_query(query):
    with engine.connect() as connection:
        result = connection.execute(text(query))
        return [dict(row) for row in result]

# Create a tool that executes SQL queries
sql_tool = Tool(
    name="SQLDatabaseTool",
    func=run_sql_query,
    description="Tool for executing SQL queries on the database."
)

# Configure the conversational agent's memory
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

# Function to handle user queries and process them with the SQL tool
def query_fn(input_text, file):
    db = process_file(file)
    conversational_agent = create_sql_agent(
        llm=llm,
        db=db,
        tools=[sql_tool],
        memory=memory,
        verbose=True,
        dialect='ansi',
        early_stopping_method="generate",
        handle_parsing_errors=True,
    )
    response = conversational_agent.run(input=input_text)

    # Check if the response is a list and contains valid data
    if isinstance(response, list) and len(response) > 0:
        # Convert each row of the result into a string
        result_text = ', '.join([str(row) for row in response])
    elif isinstance(response, str):
        # If the response is a string, use it directly
        result_text = response
    else:
        result_text = "No data found for the query."

    return {"result": result_text}

# Set up the user interface with Gradio
iface = gr.Interface(
    fn=query_fn,
    inputs=[gr.Textbox(label="Enter your query"), gr.File(label="Upload Excel file")],
    outputs=gr.JSON(label="Query Result"),
    title="Domain-specific chatbot"
)

# Launch the application
iface.launch(share=False, server_port=8080)

Create a file called "loader"

import pandas as pd
from sqlalchemy import create_engine, Table, Column, Integer, String, Date, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from langchain_community.utilities import SQLDatabase

def create_demand_table(engine, table_name, excel_file):
    # Read the Excel file
    df = pd.read_excel(excel_file)

    # Create a base class for the table model
    Base = declarative_base()
    metadata = MetaData()

    # Map pandas data types to SQLAlchemy data types
    dtype_mapping = {
        'object': String,
        'int64': Integer,
        'float64': Integer,
        'datetime64[ns]': Date,
    }

    # Dynamically define the table structure
    columns = []
    if 'ID' not in df.columns:
        columns.append(Column('id', Integer, primary_key=True, autoincrement=True))

    for col_name, dtype in df.dtypes.items():
        col_type = dtype_mapping.get(str(dtype), String)
        columns.append(Column(col_name, col_type))

    # Dynamically create the table
    demand_table = Table(table_name, metadata, *columns)

    # Drop the existing table in the SQLite database, if it exists
    metadata.drop_all(engine, [demand_table])

    # Create the table in the SQLite database
    metadata.create_all(engine)

    # Create a session to interact with the database
    Session = sessionmaker(bind=engine)

    with Session() as session:
        # Insert data into the table
        df.to_sql(table_name, con=engine, if_exists='append', index=False)

    db = SQLDatabase(engine)
    return db

Running the Application

To run the application use the following code.

python app.py

This command will start the server, and you can interact with the chatbot through the provided Gradio interface. In the console, you will see the URL where you can access your chatbot.

Gradio Interface

The generated interface consists of three parts:

Query Input Box: This is where you enter your queries in natural language.
File Upload Box: This is where you upload your Excel files.
Result Display Area: This is where the results of your queries will be displayed.

Conclusion:

AI in chatbots enhances user interactions with personalized responses and advanced natural language understanding, ensuring more engaging and contextually relevant conversations.

Deploying Bokeh Dashboard with Python on VPS: A Step-by-Step Guide

JhonnyARM — Sat, 20 Apr 2024 03:50:43 +0000

Bokeh

Bokeh is an open-source Python library for creating interactive, web-based visualizations and dashboards. It provides high-performance statistical and scientific computing capabilities, making it particularly useful for data analysis and exploration.

Interactive Plots: Bokeh allows you to create highly customizable and interactive plots, such as line plots, scatter plots, bar charts, and more.
Dashboard Applications: With Bokeh, you can build complex dashboard applications that combine multiple plots, widgets, and layouts in a single web application.
Data Visualization: Bokeh is widely used for data visualization tasks, enabling you to explore and present data in an interactive and visually appealing manner.
Scientific Computing: Bokeh integrates well with other Python libraries like NumPy, Pandas, and SciPy, making it a powerful tool for scientific computing and data analysis.
Web Applications: Bokeh's ability to create web-based visualizations makes it suitable for building web applications that require interactive data exploration and presentation.
Code: A python code was developed for a dashboard with bokeh, using the panda and bokeh libraries.

Here we have the compiled code and the generated graphics.

Creating a dashboard with BOKEH in python

Requirements:

Python
Visual Studio Code ### First steps
Open visual studio code

We create a working environment and add a .py file

Important, to make this dashboard it is necessary to install the dependencies, for this we open a console in administrator mode and we execute

pip install bokeh

Once installed, run the sample code to generate graphics.

As we can see we have the panda and bokeh packages imported.

Last step

Run the project. To run the project we will open the console and paste the following code:

python -m bokeh serve --show dashboard.py

this will show the dashboard on a local page.

Deploy the project in the cloud

To deploy the project it is necessary to have a cloud service provider, in this case I used a debian VPS.

to install python in linux

apt update

apt install python3

install the environment

sudo apt install python3-venv
mkdir my_project
cd my_project
python3 -m venv my_env
pip install bokeh

Make my_env permanent:


nano ~/.bashrc

copy and past at the end:


source /ruta/a/my_env/bin/activate

in my case it was:

source /opt/dashboardpy/my_env/bin/activate

ctrl+o ENTER ctrl+x

with "source" you're activating bashrc


source ~/.bashrc

Permanent configuration of the project

finally we make the created websocket permanent, that is to say, it does not close when closing putty, now we create a nohup that will always be executed:

nohup python -m bokeh serve --show dashboard.py &

TEST

JhonnyARM — Fri, 19 Apr 2024 06:01:11 +0000