DEV Community: César Fabián CHÁVEZ LINARES

Building Real-Time Data Pipelines with Debezium and Kafka: A Practical Guide

César Fabián CHÁVEZ LINARES — Sat, 14 Dec 2024 03:18:31 +0000

Introduction

In today's data-driven world, maintaining synchronized data across different systems is crucial. Change Data Capture (CDC) has emerged as a powerful pattern for tracking and propagating changes from your database in real-time. In this guide, we'll build a practical example using Debezium and Apache Kafka to create a robust CDC pipeline.

What We'll Build

We'll create a simple e-commerce scenario where order updates in a PostgreSQL database are automatically synchronized with an Elasticsearch instance for real-time search capabilities. This setup demonstrates a common real-world use case for CDC.

Prerequisites

Docker and Docker Compose
Java 11 or higher
Maven
Git
PostgreSQL client (psql)
curl (for testing)

Architecture Overview

Our architecture consists of several components:

PostgreSQL database (source)
Debezium connector
Apache Kafka
Kafka Connect
Elasticsearch (target)
Simple Spring Boot application for testing

graph LR
    A[PostgreSQL] -->|Debezium| B[Kafka Connect]
    B -->|Events| C[Kafka]
    C -->|Sink Connector| D[Elasticsearch]
    E[Spring Boot App] -->|Writes| A
    D -->|Search| E

Implementation Steps

1. Setting Up the Environment

First, let's create our project structure:

mkdir cdc-demo
cd cdc-demo
git init

Create a docker-compose.yml file:

version: '3'
services:
  postgres:
    image: debezium/postgres:13
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_DB=inventory
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
    volumes:
      - ./postgres/init:/docker-entrypoint-initdb.d

  kafka:
    image: confluentinc/cp-kafka:7.3.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    depends_on:
      - zookeeper

  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  connect:
    image: debezium/connect:2.1
    ports:
      - "8083:8083"
    environment:
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: my_connect_configs
      OFFSET_STORAGE_TOPIC: my_connect_offsets
      STATUS_STORAGE_TOPIC: my_connect_statuses
      BOOTSTRAP_SERVERS: kafka:29092
    depends_on:
      - kafka
      - postgres

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.17.0
    ports:
      - "9200:9200"
    environment:
      - discovery.type=single-node
      - ES_JAVA_OPTS=-Xms512m -Xmx512m

2. Creating the Database Schema

Create postgres/init/init.sql:

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    order_date TIMESTAMP NOT NULL,
    status VARCHAR(50) NOT NULL,
    total_amount DECIMAL(10,2) NOT NULL
);

ALTER TABLE orders REPLICA IDENTITY FULL;

3. Configuring Debezium

After starting the containers, configure the Debezium connector:

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
    "name": "inventory-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname": "postgres",
        "database.port": "5432",
        "database.user": "postgres",
        "database.password": "postgres",
        "database.dbname": "inventory",
        "database.server.name": "dbserver1",
        "table.include.list": "public.orders",
        "plugin.name": "pgoutput"
    }
}'

4. Spring Boot Application

Create a new Spring Boot project with the following dependencies:

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
</dependencies>

Create the Order entity:

@Entity
@Table(name = "orders")
public class Order {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private Long customerId;
    private LocalDateTime orderDate;
    private String status;
    private BigDecimal totalAmount;

    // Getters, setters, and constructors
}

Create a REST controller for testing:

@RestController
@RequestMapping("/api/orders")
public class OrderController {
    private final OrderRepository orderRepository;

    public OrderController(OrderRepository orderRepository) {
        this.orderRepository = orderRepository;
    }

    @PostMapping
    public Order createOrder(@RequestBody Order order) {
        order.setOrderDate(LocalDateTime.now());
        return orderRepository.save(order);
    }

    @PutMapping("/{id}")
    public Order updateOrder(@PathVariable Long id, @RequestBody Order order) {
        return orderRepository.findById(id)
            .map(existingOrder -> {
                existingOrder.setStatus(order.getStatus());
                existingOrder.setTotalAmount(order.getTotalAmount());
                return orderRepository.save(existingOrder);
            })
            .orElseThrow(() -> new ResponseStatusException(HttpStatus.NOT_FOUND));
    }
}

5. Testing the Pipeline

Start all containers:

docker-compose up -d

Create a test order:

curl -X POST http://localhost:8080/api/orders \
  -H "Content-Type: application/json" \
  -d '{
    "customerId": 1,
    "status": "NEW",
    "totalAmount": 99.99
  }'

Update the order:

curl -X PUT http://localhost:8080/api/orders/1 \
  -H "Content-Type: application/json" \
  -d '{
    "status": "PROCESSING",
    "totalAmount": 99.99
  }'

Check Kafka topics to verify the CDC events:

docker-compose exec kafka kafka-console-consumer \
  --bootstrap-server kafka:29092 \
  --topic dbserver1.public.orders \
  --from-beginning

Common Challenges and Solutions

Data Consistency
- Use transaction logs for accurate change capture
- Implement idempotent consumers
- Handle out-of-order events
Performance Optimization
- Batch updates when possible
- Monitor Kafka partition lag
- Tune PostgreSQL replication slots
Error Handling
- Implement dead letter queues
- Set up proper monitoring and alerting
- Create retry mechanisms

Best Practices

Schema Evolution
- Use Avro for schema management
- Plan for backward/forward compatibility
- Test schema changes thoroughly
Monitoring
- Track replication lag
- Monitor Kafka consumer group offsets
- Set up alerts for connector failures
Security
- Use SSL/TLS for communication
- Implement proper authentication
- Follow least privilege principle

Conclusion

CDC with Debezium and Kafka provides a robust solution for real-time data synchronization. This setup can be extended to handle more complex scenarios like:

Multi-region deployment
Multiple target systems
Complex transformation pipelines
High availability requirements

Resources

Building a Real-time Chat Application with Google Cloud Firestore and React

César Fabián CHÁVEZ LINARES — Tue, 10 Dec 2024 17:43:01 +0000

In this tutorial, we'll create a real-time chat application using Google Cloud Firestore and React. Firestore offers powerful real-time synchronization capabilities, making it perfect for chat applications, collaborative tools, and live dashboards.

Why Google Cloud Firestore?

While DynamoDB and CosmosDB are popular choices, Firestore offers unique advantages:

Real-time listeners with automatic data synchronization
Powerful querying capabilities
Offline data persistence
Automatic scaling
Simple SDK integration
Free tier for development and small applications

Prerequisites

Node.js and npm installed
Google Cloud account
Basic knowledge of React and JavaScript
Familiarity with async/await

Project Setup

First, create a new React project and install dependencies:

npx create-react-app chat-app
cd chat-app
npm install firebase @mui/material @mui/icons-material @emotion/react @emotion/styled date-fns

Firebase Configuration

Create a new project in the Firebase Console
Enable Firestore in your project
Create src/firebase.js:

import { initializeApp } from 'firebase/app';
import { getFirestore } from 'firebase/firestore';
import { getAuth } from 'firebase/auth';

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

const app = initializeApp(firebaseConfig);
export const db = getFirestore(app);
export const auth = getAuth(app);

Database Schema

Create the following collections in Firestore:

rooms/
  ├── roomId/
  │   ├── name: string
  │   ├── description: string
  │   └── createdAt: timestamp
  │
messages/
  ├── messageId/
  │   ├── roomId: string
  │   ├── userId: string
  │   ├── text: string
  │   ├── createdAt: timestamp
  │   └── userName: string

Components Implementation

Create the chat components:

ChatRoom Component (`src/components/ChatRoom.js`):

import React, { useEffect, useState, useRef } from 'react';
import { collection, query, orderBy, limit, onSnapshot, addDoc, serverTimestamp } from 'firebase/firestore';
import { db, auth } from '../firebase';
import { Box, TextField, Button, Paper, Typography, List, ListItem, ListItemText } from '@mui/material';
import { formatDistance } from 'date-fns';

const ChatRoom = ({ roomId }) => {
  const [messages, setMessages] = useState([]);
  const [newMessage, setNewMessage] = useState('');
  const messagesEndRef = useRef(null);

  useEffect(() => {
    // Query messages for this room
    const q = query(
      collection(db, 'messages'),
      orderBy('createdAt', 'asc'),
      limit(100)
    );

    // Subscribe to real-time updates
    const unsubscribe = onSnapshot(q, (snapshot) => {
      const newMessages = [];
      snapshot.forEach((doc) => {
        newMessages.push({ id: doc.id, ...doc.data() });
      });
      setMessages(newMessages);
      scrollToBottom();
    });

    return () => unsubscribe();
  }, [roomId]);

  const scrollToBottom = () => {
    messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' });
  };

  const handleSubmit = async (e) => {
    e.preventDefault();
    if (!newMessage.trim()) return;

    try {
      await addDoc(collection(db, 'messages'), {
        text: newMessage,
        createdAt: serverTimestamp(),
        userId: auth.currentUser.uid,
        userName: auth.currentUser.displayName || 'Anonymous',
        roomId
      });

      setNewMessage('');
    } catch (error) {
      console.error('Error sending message:', error);
    }
  };

  return (
    <Box sx={{ height: '100vh', display: 'flex', flexDirection: 'column' }}>
      <Paper elevation={3} sx={{ flex: 1, overflow: 'auto', p: 2, mb: 2 }}>
        <List>
          {messages.map((message) => (
            <ListItem
              key={message.id}
              sx={{
                display: 'flex',
                flexDirection: 'column',
                alignItems: message.userId === auth.currentUser?.uid ? 'flex-end' : 'flex-start'
              }}
            >
              <Typography variant="caption" color="textSecondary">
                {message.userName} • {
                  message.createdAt && formatDistance(message.createdAt.toDate(), new Date(), { addSuffix: true })
                }
              </Typography>
              <Paper
                elevation={1}
                sx={{
                  p: 1,
                  bgcolor: message.userId === auth.currentUser?.uid ? 'primary.light' : 'grey.100',
                  maxWidth: '70%'
                }}
              >
                <Typography>{message.text}</Typography>
              </Paper>
            </ListItem>
          ))}
          <div ref={messagesEndRef} />
        </List>
      </Paper>

      <Box component="form" onSubmit={handleSubmit} sx={{ p: 2 }}>
        <TextField
          fullWidth
          value={newMessage}
          onChange={(e) => setNewMessage(e.target.value)}
          placeholder="Type a message..."
          variant="outlined"
          sx={{ mr: 1 }}
          InputProps={{
            endAdornment: (
              <Button type="submit" variant="contained" disabled={!newMessage.trim()}>
                Send
              </Button>
            )
          }}
        />
      </Box>
    </Box>
  );
};

export default ChatRoom;

Room List Component (`src/components/RoomList.js`):

import React, { useEffect, useState } from 'react';
import { collection, query, orderBy, onSnapshot, addDoc, serverTimestamp } from 'firebase/firestore';
import { db } from '../firebase';
import { List, ListItem, ListItemText, Button, Dialog, DialogTitle, 
         DialogContent, TextField, DialogActions } from '@mui/material';

const RoomList = ({ onRoomSelect }) => {
  const [rooms, setRooms] = useState([]);
  const [open, setOpen] = useState(false);
  const [newRoomName, setNewRoomName] = useState('');
  const [newRoomDescription, setNewRoomDescription] = useState('');

  useEffect(() => {
    const q = query(collection(db, 'rooms'), orderBy('createdAt', 'desc'));

    const unsubscribe = onSnapshot(q, (snapshot) => {
      const newRooms = [];
      snapshot.forEach((doc) => {
        newRooms.push({ id: doc.id, ...doc.data() });
      });
      setRooms(newRooms);
    });

    return () => unsubscribe();
  }, []);

  const handleCreateRoom = async () => {
    if (!newRoomName.trim()) return;

    try {
      await addDoc(collection(db, 'rooms'), {
        name: newRoomName,
        description: newRoomDescription,
        createdAt: serverTimestamp()
      });

      setOpen(false);
      setNewRoomName('');
      setNewRoomDescription('');
    } catch (error) {
      console.error('Error creating room:', error);
    }
  };

  return (
    <>
      <List>
        {rooms.map((room) => (
          <ListItem
            key={room.id}
            button
            onClick={() => onRoomSelect(room.id)}
          >
            <ListItemText 
              primary={room.name}
              secondary={room.description}
            />
          </ListItem>
        ))}
      </List>

      <Button variant="contained" onClick={() => setOpen(true)}>
        Create New Room
      </Button>

      <Dialog open={open} onClose={() => setOpen(false)}>
        <DialogTitle>Create New Chat Room</DialogTitle>
        <DialogContent>
          <TextField
            autoFocus
            margin="dense"
            label="Room Name"
            fullWidth
            value={newRoomName}
            onChange={(e) => setNewRoomName(e.target.value)}
          />
          <TextField
            margin="dense"
            label="Description"
            fullWidth
            multiline
            rows={2}
            value={newRoomDescription}
            onChange={(e) => setNewRoomDescription(e.target.value)}
          />
        </DialogContent>
        <DialogActions>
          <Button onClick={() => setOpen(false)}>Cancel</Button>
          <Button onClick={handleCreateRoom} variant="contained">
            Create
          </Button>
        </DialogActions>
      </Dialog>
    </>
  );
};

export default RoomList;

Main App Component (`src/App.js`):

import React, { useState } from 'react';
import { Container, Grid, Paper } from '@mui/material';
import ChatRoom from './components/ChatRoom';
import RoomList from './components/RoomList';

function App() {
  const [selectedRoom, setSelectedRoom] = useState(null);

  return (
    <Container maxWidth="lg" sx={{ height: '100vh', py: 2 }}>
      <Grid container spacing={2} sx={{ height: '100%' }}>
        <Grid item xs={3}>
          <Paper sx={{ height: '100%', p: 2 }}>
            <RoomList onRoomSelect={setSelectedRoom} />
          </Paper>
        </Grid>
        <Grid item xs={9}>
          {selectedRoom ? (
            <ChatRoom roomId={selectedRoom} />
          ) : (
            <Paper sx={{ height: '100%', display: 'flex', alignItems: 'center', justifyContent: 'center' }}>
              Select a room to start chatting
            </Paper>
          )}
        </Grid>
      </Grid>
    </Container>
  );
}

export default App;

Security Rules

Set up Firestore security rules:

rules_version = '2';
service cloud.firestore {
  match /databases/{database}/documents {
    match /messages/{messageId} {
      allow read: if true;
      allow create: if request.auth != null;
      allow update, delete: if request.auth != null && 
                            request.auth.uid == resource.data.userId;
    }

    match /rooms/{roomId} {
      allow read: if true;
      allow create: if request.auth != null;
      allow update, delete: if request.auth != null;
    }
  }
}

Running the Application

Set up environment variables:

cp .env.example .env
# Add your Firebase configuration

Start the development server:

npm start

Key Features

Real-time message synchronization
Room creation and management
Message history persistence
User authentication integration
Responsive Material-UI design
Automatic message timestamps
Scroll-to-bottom functionality

Performance Considerations

When working with Firestore:

Query Optimization
- Use compound queries for better performance
- Implement pagination for large datasets
- Cache frequently accessed data
Real-time Listeners
- Detach listeners when components unmount
- Limit the number of concurrent listeners
- Use appropriate query limits
Offline Support
- Enable offline persistence for better user experience
- Handle offline/online state transitions
- Implement retry logic for failed operations
Cost Optimization
- Monitor read/write operations
- Use batch operations when possible
- Implement appropriate caching strategies

Deployment

Build the application:

npm run build

Deploy to Firebase Hosting:

npm install -g firebase-tools
firebase login
firebase init
firebase deploy

Conclusion

This implementation showcases how to build a real-time chat application using Google Cloud Firestore. The combination of Firestore's real-time capabilities and React's component model makes it easy to create responsive, scalable applications.

Future enhancements could include:

File attachments
User presence indicators
Message reactions
Rich text formatting
Voice messages

The complete source code is available on GitHub: [Link to your repository]

Resources

Building a Task Management API with Apache Cassandra and FastAPI

César Fabián CHÁVEZ LINARES — Tue, 10 Dec 2024 17:39:00 +0000

In this tutorial, we'll create a scalable task management API using Apache Cassandra as our NoSQL database and FastAPI for the web framework. Cassandra is an excellent choice for applications requiring high availability, horizontal scalability, and consistent performance with large datasets.

Why Apache Cassandra?

While MongoDB and Redis are popular NoSQL choices, Cassandra offers unique benefits:

Linear scalability with no single point of failure
Tunable consistency levels
Optimized for write-heavy workloads
Excellent support for time-series data
Built-in partitioning and replication

Prerequisites

Python 3.8+
Docker and Docker Compose
Basic understanding of REST APIs
Familiarity with async programming

Project Setup

First, let's create our project structure:

task-manager/
├── docker-compose.yml
├── requirements.txt
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── database.py
│   └── routes/
│       └── tasks.py
└── README.md

Let's set up our dependencies in requirements.txt:

fastapi==0.104.1
uvicorn==0.24.0
cassandra-driver==3.28.0
pydantic==2.4.2
python-dotenv==1.0.0

Create a docker-compose.yml for Cassandra:

version: '3'

services:
  cassandra:
    image: cassandra:latest
    ports:
      - "9042:9042"
    environment:
      - CASSANDRA_CLUSTER_NAME=TaskManagerCluster
    volumes:
      - cassandra_data:/var/lib/cassandra

volumes:
  cassandra_data:

Database Setup

Let's create our database connection handler in database.py:

from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from cassandra.cqlengine import connection
from contextlib import asynccontextmanager

class CassandraConnection:
    def __init__(self):
        self.cluster = None
        self.session = None

    async def connect(self):
        self.cluster = Cluster(['localhost'])
        self.session = self.cluster.connect()

        # Create keyspace if not exists
        self.session.execute("""
            CREATE KEYSPACE IF NOT EXISTS task_manager
            WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}
        """)

        self.session.set_keyspace('task_manager')

        # Create tasks table
        self.session.execute("""
            CREATE TABLE IF NOT EXISTS tasks (
                task_id uuid,
                title text,
                description text,
                status text,
                due_date timestamp,
                created_at timestamp,
                updated_at timestamp,
                PRIMARY KEY (task_id)
            )
        """)

        return self.session

    async def disconnect(self):
        if self.cluster:
            self.cluster.shutdown()

db = CassandraConnection()

@asynccontextmanager
async def get_db():
    try:
        session = await db.connect()
        yield session
    finally:
        await db.disconnect()

Models

Define our data models in models.py:

from datetime import datetime
from uuid import UUID, uuid4
from pydantic import BaseModel, Field
from typing import Optional

class TaskBase(BaseModel):
    title: str
    description: Optional[str] = None
    status: str = "pending"
    due_date: Optional[datetime] = None

class TaskCreate(TaskBase):
    pass

class Task(TaskBase):
    task_id: UUID = Field(default_factory=uuid4)
    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: datetime = Field(default_factory=datetime.now)

    class Config:
        from_attributes = True

API Routes

Create our task routes in routes/tasks.py:

from fastapi import APIRouter, Depends, HTTPException
from cassandra.cluster import Session
from datetime import datetime
import uuid
from typing import List
from ..models import Task, TaskCreate
from ..database import get_db

router = APIRouter()

@router.post("/tasks/", response_model=Task)
async def create_task(task: TaskCreate, db: Session = Depends(get_db)):
    task_id = uuid.uuid4()
    now = datetime.now()

    query = """
        INSERT INTO tasks (task_id, title, description, status, due_date, created_at, updated_at)
        VALUES (%s, %s, %s, %s, %s, %s, %s)
    """

    db.execute(query, (
        task_id,
        task.title,
        task.description,
        task.status,
        task.due_date,
        now,
        now
    ))

    return Task(
        task_id=task_id,
        title=task.title,
        description=task.description,
        status=task.status,
        due_date=task.due_date,
        created_at=now,
        updated_at=now
    )

@router.get("/tasks/{task_id}", response_model=Task)
async def read_task(task_id: uuid.UUID, db: Session = Depends(get_db)):
    query = "SELECT * FROM tasks WHERE task_id = %s"
    result = db.execute(query, [task_id]).one()

    if not result:
        raise HTTPException(status_code=404, detail="Task not found")

    return Task(**result)

@router.get("/tasks/", response_model=List[Task])
async def list_tasks(db: Session = Depends(get_db)):
    query = "SELECT * FROM tasks"
    results = db.execute(query)
    return [Task(**row) for row in results]

@router.put("/tasks/{task_id}", response_model=Task)
async def update_task(task_id: uuid.UUID, task: TaskCreate, db: Session = Depends(get_db)):
    # Check if task exists
    exists_query = "SELECT * FROM tasks WHERE task_id = %s"
    existing_task = db.execute(exists_query, [task_id]).one()

    if not existing_task:
        raise HTTPException(status_code=404, detail="Task not found")

    update_query = """
        UPDATE tasks 
        SET title = %s, description = %s, status = %s, due_date = %s, updated_at = %s
        WHERE task_id = %s
    """

    now = datetime.now()
    db.execute(update_query, (
        task.title,
        task.description,
        task.status,
        task.due_date,
        now,
        task_id
    ))

    return Task(
        task_id=task_id,
        title=task.title,
        description=task.description,
        status=task.status,
        due_date=task.due_date,
        created_at=existing_task.created_at,
        updated_at=now
    )

@router.delete("/tasks/{task_id}")
async def delete_task(task_id: uuid.UUID, db: Session = Depends(get_db)):
    # Check if task exists
    exists_query = "SELECT * FROM tasks WHERE task_id = %s"
    existing_task = db.execute(exists_query, [task_id]).one()

    if not existing_task:
        raise HTTPException(status_code=404, detail="Task not found")

    query = "DELETE FROM tasks WHERE task_id = %s"
    db.execute(query, [task_id])

    return {"message": "Task deleted successfully"}

Main Application

Set up our FastAPI application in main.py:

from fastapi import FastAPI
from .routes import tasks

app = FastAPI(title="Task Manager API")

app.include_router(tasks.router, tags=["tasks"])

@app.get("/")
async def root():
    return {"message": "Welcome to Task Manager API"}

Running the Application

Start Cassandra:

docker-compose up -d

Install dependencies:

pip install -r requirements.txt

Run the API:

uvicorn app.main:app --reload

Testing the API

Here are some example curl commands to test our API:

# Create a task
curl -X POST "http://localhost:8000/tasks/" \
     -H "Content-Type: application/json" \
     -d '{"title":"Complete project","description":"Finish the API implementation","status":"pending"}'

# List all tasks
curl "http://localhost:8000/tasks/"

# Get a specific task
curl "http://localhost:8000/tasks/{task_id}"

# Update a task
curl -X PUT "http://localhost:8000/tasks/{task_id}" \
     -H "Content-Type: application/json" \
     -d '{"title":"Complete project","description":"API implementation done","status":"completed"}'

# Delete a task
curl -X DELETE "http://localhost:8000/tasks/{task_id}"

Performance Considerations

When working with Cassandra, keep in mind:

Data Modeling: Design tables around your query patterns
Partition Keys: Choose partition keys that distribute data evenly
Consistency Levels: Configure based on your use case
Batch Operations: Use them sparingly and only for related data
Secondary Indexes: Use them carefully as they can impact performance

Conclusion

This implementation demonstrates how to build a robust task management API using Apache Cassandra. The combination of Cassandra's scalability and FastAPI's performance makes this stack suitable for production applications requiring high availability and consistency.

Future enhancements could include:

Authentication and authorization
More complex querying capabilities
Batch operations
Advanced task filtering
Automated testing

The complete source code is available on GitHub: [Link to your repository]

Resources

Building a Simple Generative AI Chatbot: A Practical Guide

César Fabián CHÁVEZ LINARES — Tue, 10 Dec 2024 17:32:42 +0000

In this tutorial, we'll walk through creating a generative AI chatbot using Python and the OpenAI API. We'll build a chatbot that can engage in natural conversations while maintaining context and providing helpful responses.

Prerequisites

Python 3.8+
Basic understanding of Python programming
OpenAI API key
Basic knowledge of RESTful APIs

Setting Up the Environment

First, let's set up our development environment. Create a new Python project and install the required dependencies:

pip install openai python-dotenv streamlit

Project Structure

Our chatbot will have a clean, modular structure:

chatbot/
├── .env
├── app.py
├── chat_handler.py
└── requirements.txt

Implementation

Let's start with our core chatbot logic in chat_handler.py:

import openai
from typing import List, Dict
import os
from dotenv import load_dotenv

load_dotenv()

class ChatBot:
    def __init__(self):
        openai.api_key = os.getenv("OPENAI_API_KEY")
        self.conversation_history: List[Dict[str, str]] = []
        self.system_prompt = """You are a helpful AI assistant. Provide clear, 
        accurate, and engaging responses while maintaining a friendly tone."""

    def add_message(self, role: str, content: str):
        self.conversation_history.append({"role": role, "content": content})

    def get_response(self, user_input: str) -> str:
        # Add user input to conversation history
        self.add_message("user", user_input)

        # Prepare messages for API call
        messages = [{"role": "system", "content": self.system_prompt}] + \
                  self.conversation_history

        try:
            # Make API call to OpenAI
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=messages,
                max_tokens=1000,
                temperature=0.7
            )

            # Extract and store assistant's response
            assistant_response = response.choices[0].message.content
            self.add_message("assistant", assistant_response)

            return assistant_response

        except Exception as e:
            return f"An error occurred: {str(e)}"

Now, let's create a simple web interface using Streamlit in app.py:

import streamlit as st
from chat_handler import ChatBot

def main():
    st.title("🤖 AI Chatbot")

    # Initialize session state
    if "chatbot" not in st.session_state:
        st.session_state.chatbot = ChatBot()

    # Chat interface
    if "messages" not in st.session_state:
        st.session_state.messages = []

    # Display chat history
    for message in st.session_state.messages:
        with st.chat_message(message["role"]):
            st.write(message["content"])

    # Chat input
    if prompt := st.chat_input("What's on your mind?"):
        # Add user message to chat history
        st.session_state.messages.append({"role": "user", "content": prompt})
        with st.chat_message("user"):
            st.write(prompt)

        # Get bot response
        response = st.session_state.chatbot.get_response(prompt)

        # Add assistant response to chat history
        st.session_state.messages.append({"role": "assistant", "content": response})
        with st.chat_message("assistant"):
            st.write(response)

if __name__ == "__main__":
    main()

Key Features

Conversation Memory: The chatbot maintains context by storing the conversation history.
System Prompt: We define the chatbot's behavior and personality through a system prompt.
Error Handling: The implementation includes basic error handling for API calls.
User Interface: A clean, intuitive web interface using Streamlit.

Running the Chatbot

Create a .env file with your OpenAI API key:

OPENAI_API_KEY=your_api_key_here

Run the application:

streamlit run app.py

Potential Enhancements

Conversation Persistence: Add database integration to store chat histories.
Custom Personalities: Allow users to select different chatbot personalities.
Input Validation: Add more robust input validation and sanitization.
API Rate Limiting: Implement rate limiting to manage API usage.
Response Streaming: Add streaming responses for better user experience.

Conclusion

This implementation demonstrates a basic but functional generative AI chatbot. The modular design makes it easy to extend and customize based on specific needs. While this example uses OpenAI's API, the same principles can be applied with other language models or APIs.

Remember that when deploying a chatbot, you should consider:

API costs and usage limits
User data privacy and security
Response latency and optimization
Input validation and content moderation

Resources

Creating a Dashboard with Streamlit

César Fabián CHÁVEZ LINARES — Sat, 31 Aug 2024 16:54:46 +0000

Introduction
In this document, we will explore how to create an interactive dashboard using Streamlit, an open-source tool for building web applications in Python. Streamlit simplifies the development process by allowing you to create web interfaces with just a few lines of code.

Requirements

To follow this example, you will need to have Python installed and the following libraries:

streamlit
pandas
numpy
altair

You can install them using pip:

pip install streamlit pandas numpy altair

Code Structure

The code is divided into several sections:

Library imports: We import the necessary libraries to generate sample data and create visualizations.
Generation of sample data: We create a Pandas DataFrame with random data to use in our visualizations.
Dashboard setup: We set the title of the dashboard using st.title().
Creation of visualizations:
- We display a line chart using st.line_chart().
- We create an interactive scatter plot using Altair and display it with st.altair_chart().

Code Explanation

Library imports:

streamlit as st: We import the Streamlit library and rename it as st for easier use.
pandas as pd: We import Pandas and rename it as pd.
numpy as np: We import NumPy and rename it as np.
altair as alt: We import Altair and rename it as alt.

Generation of sample data:

We create a Pandas DataFrame called data with two columns: 'x' and 'y'.
The 'x' column contains integer values from 0 to 99 using np.arange().
The 'y' column contains random values between 0 and 1 using np.random.rand().

Dashboard setup:

We set the title of the dashboard using st.title() and passing the title as an argument.

Creation of visualizations:

We display a line chart using st.line_chart() and passing the data DataFrame as an argument.
We create an interactive scatter plot using Altair:

alt.Chart(data): We create a chart object from the data DataFrame.
.mark_circle(): We specify that we want to use circles as marks.
.encode(x='x', y='y', tooltip=['x', 'y']): We assign the 'x' and 'y' columns to the x and y axes respectively, and specify that we want to show the 'x' and 'y' columns as tooltips.
.interactive(): Makes the chart interactive, allowing the user to zoom and pan.

We display the Altair chart using st.altair_chart() and pass the chart object as an argument. We also specify use_container_width=True so that the chart takes up the full width of the container.

Deployment on Streamlit Cloud

To deploy this Streamlit app, you can use Streamlit Cloud:

Upload your code to a GitHub repository.
Sign in to Streamlit Cloud.
Connect your GitHub repository.
Deploy your app with a few clicks.

Streamlit Cloud handles the entire deployment process, making it very easy to share your applications with the world.

Results

Repository url: [](https://github.com/FabianChavezLinares/Research-Group-Activity.git)

Page url: https://research-group-activity-bi-2024-ii.streamlit.app

Conclusion

In this document, We have learned how to create an interactive dashboard using Streamlit. We generated sample data, created visualizations using Streamlit and Altair, and explained each section of the code. Finally, We mentioned how to deploy the application on Streamlit Cloud to share it with other users.

Streamlit greatly simplifies the process of creating interactive web applications in Python, making it a valuable tool for data scientists and developers who want to share their visualizations and analyses in an easy and quick manner.

Creating a Simple Generative AI Chatbot with Python and TensorFlow

César Fabián CHÁVEZ LINARES — Tue, 09 Jul 2024 22:33:39 +0000

Introduction:
In this article, we'll walk through the process of creating a basic generative AI chatbot using Python and TensorFlow. This chatbot will be capable of generating responses based on input text, showcasing the fundamentals of natural language processing and neural networks.
Prerequisites:

Basic knowledge of Python
Familiarity with neural networks concepts
Python 3.7 or later installed
TensorFlow 2.x installed

Step 1: Setting Up the Environment
First, ensure you have the necessary libraries installed:

Step 2: Preparing the Data
We'll use a simple dataset for this example. Create a file named 'conversations.txt' with some sample dialogue:

Step 3: Preprocessing the Data
Now, let's preprocess our data:

Step 4: Building the Model
Let's create a simple sequence-to-sequence model:

Step 5: Training the Model
Now, let's train our model:

Step 6: Using the Chatbot
Finally, let's create a function to generate responses:

Conclusion:
This article demonstrates how to create a simple generative AI chatbot using Python and TensorFlow. While this example is basic, it provides a foundation for more complex chatbot implementations. Future improvements could include using larger datasets, implementing attention mechanisms, or exploring more advanced architectures like transformers.

Remember, creating AI models requires ethical considerations and responsible use. Always ensure your chatbot is designed to be helpful and not harmful.

DEV Community: César Fabián CHÁVEZ LINARES

Building Real-Time Data Pipelines with Debezium and Kafka: A Practical Guide

Introduction

What We'll Build

Prerequisites

Architecture Overview

Implementation Steps

1. Setting Up the Environment

2. Creating the Database Schema

3. Configuring Debezium

4. Spring Boot Application

5. Testing the Pipeline

Common Challenges and Solutions

Best Practices

Conclusion

Resources

Building a Real-time Chat Application with Google Cloud Firestore and React

Why Google Cloud Firestore?

Prerequisites

Project Setup

Firebase Configuration

Database Schema

Components Implementation

ChatRoom Component (src/components/ChatRoom.js):

Room List Component (src/components/RoomList.js):

Main App Component (src/App.js):

Security Rules

Running the Application

Key Features

Performance Considerations

Deployment

Conclusion

Resources

Building a Task Management API with Apache Cassandra and FastAPI

Why Apache Cassandra?

Prerequisites

Project Setup

Database Setup

Models

API Routes

Main Application

Running the Application

Testing the API

Performance Considerations

Conclusion

Resources

Building a Simple Generative AI Chatbot: A Practical Guide

Prerequisites

Setting Up the Environment

Project Structure

Implementation

Key Features

Running the Chatbot

Potential Enhancements

Conclusion

Resources

Creating a Dashboard with Streamlit

Creating a Simple Generative AI Chatbot with Python and TensorFlow

ChatRoom Component (`src/components/ChatRoom.js`):

Room List Component (`src/components/RoomList.js`):

Main App Component (`src/App.js`):