🗄️ NoSQL Fundamentals
🎯 What is NoSQL?
NoSQL (Not Only SQL) refers to a class of databases that do not use the relational model with fixed schemas. They are designed for scalability, flexibility, and handling large volumes of diverse data.
Key concept: NoSQL databases are schema-less or schema-flexible, supporting horizontal scaling across clusters of commodity hardware.
Deep Explanation
NoSQL emerged to address the limitations of relational databases in handling big data, high velocity, and varied data types. They are optimized for specific data models and access patterns.
Why it matters: NoSQL powers web-scale applications, real-time analytics, IoT, and modern cloud-native systems.
# NoSQL database types (conceptual)
class NoSQLDatabase:
def __init__(self, type, config):
self.type = type
self.config = config
self.data = {}
def insert(self, key, value):
self.data[key] = value
def get(self, key):
return self.data.get(key)
def delete(self, key):
if key in self.data:
del self.data[key]
# Example: Document database (MongoDB)
mongo = NoSQLDatabase('document', {'host': 'localhost', 'port': 27017})
mongo.insert('user_001', {'name': 'Alice', 'age': 30, 'city': 'NYC'})
print(mongo.get('user_001'))
# Example: Key-value store (Redis)
redis = NoSQLDatabase('key-value', {'host': 'localhost', 'port': 6379})
redis.insert('session:abc123', {'user_id': 123, 'expires': '2024-01-01'})📊 Database Types
🎯 Four Main NoSQL Types
NoSQL databases are categorized into four main types: Key-Value, Document, Column-Family, and Graph databases.
Type Comparison
Key-Value: Simple, fast, uses hash tables (Redis, DynamoDB). Document: JSON-like documents with nested structures (MongoDB, CouchDB). Column-Family: Wide-column stores (Cassandra, HBase). Graph: Nodes and edges for relationships (Neo4j, Amazon Neptune).
Choice factors: Data model, access patterns, scalability needs.
# Database type examples
class KeyValueStore:
def __init__(self):
self.store = {}
def put(self, key, value):
self.store[key] = value
def get(self, key):
return self.store.get(key)
class DocumentStore:
def __init__(self):
self.store = {}
def insert(self, doc_id, document):
self.store[doc_id] = document
def find(self, query):
# Find documents matching query
results = []
for doc_id, doc in self.store.items():
if all(doc.get(k) == v for k, v in query.items()):
results.append((doc_id, doc))
return results
class ColumnFamilyStore:
def __init__(self):
self.families = {}
def create_family(self, name, columns):
self.families[name] = {'columns': columns, 'rows': {}}
def insert(self, family, key, data):
self.families[family]['rows'][key] = data
class GraphDatabase:
def __init__(self):
self.nodes = {}
self.edges = []
def add_node(self, id, label, properties):
self.nodes[id] = {'label': label, 'properties': properties}
def add_edge(self, from_id, to_id, type, properties):
self.edges.append({'from': from_id, 'to': to_id, 'type': type, 'properties': properties})🔑 Key-Value Stores
🎯 The Simplest NoSQL
Key-Value stores are the simplest NoSQL databases, using a hash table where each key maps to a value. They offer O(1) access for read/write operations.
Key-Value Features
Examples: Redis, Amazon DynamoDB, Riak, Memcached. Use cases: Caching, session management, real-time bidding, shopping carts. Performance: Extremely fast, sub-millisecond latency.
Limitations: No query language, limited to key-based access.
# Redis (Python)
import redis
# Connect to Redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
# Set and get values
r.set('user:1001', '{"name": "Alice", "age": 30}')
user = r.get('user:1001')
print(user)
# Expiring keys
r.setex('session:abc123', 3600, 'user_id:1001') # Expires in 1 hour
# List operations
r.rpush('tasks', 'task1', 'task2', 'task3')
tasks = r.lrange('tasks', 0, -1)
print(tasks)
# Atomic operations
r.incr('counter') # Increment counter
r.incrby('counter', 5) # Increment by 5
# Hash operations
r.hset('user:1002', 'name', 'Bob')
r.hset('user:1002', 'age', 25)
r.hset('user:1002', 'email', 'bob@email.com')
user_data = r.hgetall('user:1002')
print(user_data)
# Set operations
r.sadd('tags:python', 'web', 'data', 'ml')
r.sadd('tags:java', 'web', 'enterprise')
common = r.sinter('tags:python', 'tags:java') # Intersection
print(common)📄 Document Databases
🎯 Flexible, JSON-Like Documents
Document databases store data in JSON-like documents with nested structures. They provide flexible schemas and powerful query languages.
Document Features
Examples: MongoDB, CouchDB, Firebase Firestore. Use cases: Content management, catalogs, user profiles, real-time apps. Key features: Rich queries, indexing, aggregation pipelines.
Data model: BSON/JSON with nested objects and arrays.
# MongoDB (PyMongo)
from pymongo import MongoClient
# Connect to MongoDB
client = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['users']
# Insert a document
user = {
'name': 'Alice',
'age': 30,
'email': 'alice@email.com',
'address': {
'city': 'New York',
'state': 'NY',
'zip': '10001'
},
'hobbies': ['reading', 'hiking', 'coding']
}
result = collection.insert_one(user)
print(f"Inserted ID: {result.inserted_id}")
# Insert multiple documents
users = [
{'name': 'Bob', 'age': 25, 'email': 'bob@email.com'},
{'name': 'Charlie', 'age': 35, 'email': 'charlie@email.com'}
]
collection.insert_many(users)
# Query documents
user = collection.find_one({'name': 'Alice'})
print(user)
# Find all users
all_users = collection.find({})
for u in all_users:
print(u['name'])
# Query with conditions
young_users = collection.find({'age': {'$lt': 30}})
for u in young_users:
print(f"{u['name']}: {u['age']}")
# Update document
collection.update_one(
{'name': 'Alice'},
{'$set': {'age': 31, 'hobbies': ['reading', 'hiking', 'coding', 'writing']}}
)
# Aggregation pipeline
pipeline = [
{'$match': {'age': {'$gte': 25}}},
{'$group': {'_id': '$city', 'avg_age': {'$avg': '$age'}}},
{'$sort': {'avg_age': -1}}
]
result = collection.aggregate(pipeline)📋 Column-Family Stores
🎯 Wide-Column Databases
Column-family stores organize data into rows and columns but with a flexible schema. They are optimized for high write throughput and horizontal scaling.
Column-Family Features
Examples: Apache Cassandra, HBase, ScyllaDB. Use cases: Time-series data, IoT, logging, real-time analytics. Key features: Tunable consistency, distribution, high write performance.
Data model: Row key → Column families → Columns.
# Cassandra (Python)
from cassandra.cluster import Cluster
from cassandra.query import SimpleStatement
# Connect to Cassandra
cluster = Cluster(['localhost'])
session = cluster.connect()
# Create keyspace
session.execute("""
CREATE KEYSPACE IF NOT EXISTS demo
WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}
""")
session.set_keyspace('demo')
# Create table
session.execute("""
CREATE TABLE IF NOT EXISTS users (
user_id UUID PRIMARY KEY,
name TEXT,
age INT,
email TEXT
)
""")
# Insert data
from uuid import uuid4
session.execute(
"INSERT INTO users (user_id, name, age, email) VALUES (%s, %s, %s, %s)",
(uuid4(), 'Alice', 30, 'alice@email.com')
)
# Query data
rows = session.execute("SELECT * FROM users")
for row in rows:
print(f"User: {row.name}, Age: {row.age}")
# Create table with clustering columns
session.execute("""
CREATE TABLE IF NOT EXISTS sensor_data (
sensor_id TEXT,
timestamp TIMESTAMP,
temperature FLOAT,
humidity FLOAT,
PRIMARY KEY (sensor_id, timestamp)
) WITH CLUSTERING ORDER BY (timestamp DESC)
""")
# Insert sensor data
session.execute(
"INSERT INTO sensor_data (sensor_id, timestamp, temperature, humidity) VALUES (%s, %s, %s, %s)",
('sensor1', datetime.now(), 23.5, 65.0)
)
# Query by sensor_id (partition key)
rows = session.execute("SELECT * FROM sensor_data WHERE sensor_id = 'sensor1'")
for row in rows:
print(f"{row.timestamp}: {row.temperature}°C, {row.humidity}%")
# Batch insert (atomic)
batch = SimpleStatement(
"BEGIN BATCH "
"INSERT INTO users (user_id, name, age) VALUES (uuid(), 'Bob', 25) "
"INSERT INTO users (user_id, name, age) VALUES (uuid(), 'Charlie', 35) "
"APPLY BATCH"
)
session.execute(batch)🕸️ Graph Databases
🎯 Relationships as First-Class
Graph databases store entities as nodes and relationships as edges. They excel at traversing and querying complex relationships.
Graph Features
Examples: Neo4j, Amazon Neptune, TigerGraph. Use cases: Social networks, fraud detection, recommendation engines, knowledge graphs. Key features: Cypher query language, path traversals, real-time queries.
Data model: Nodes + Edges + Properties.
# Neo4j (Python)
from neo4j import GraphDatabase
class Neo4jConnection:
def __init__(self, uri, user, password):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
self.driver.close()
def create_graph(self):
with self.driver.session() as session:
# Create nodes
session.run("CREATE (alice:Person {name: 'Alice', age: 30})")
session.run("CREATE (bob:Person {name: 'Bob', age: 25})")
session.run("CREATE (charlie:Person {name: 'Charlie', age: 35})")
session.run("CREATE (david:Person {name: 'David', age: 28})")
# Create relationships
session.run("MATCH (a:Person {name: 'Alice'}), (b:Person {name: 'Bob'}) "
"CREATE (a)-[:KNOWS]->(b)")
session.run("MATCH (b:Person {name: 'Bob'}), (c:Person {name: 'Charlie'}) "
"CREATE (b)-[:KNOWS]->(c)")
session.run("MATCH (a:Person {name: 'Alice'}), (c:Person {name: 'Charlie'}) "
"CREATE (a)-[:KNOWS]->(c)")
def find_friends(self, name):
with self.driver.session() as session:
result = session.run(
"MATCH (p:Person {name: $name})-[:KNOWS]-(friend) "
"RETURN friend.name, friend.age",
name=name
)
return [(record['friend.name'], record['friend.age'])
for record in result]
def shortest_path(self, start, end):
with self.driver.session() as session:
result = session.run(
"MATCH path = shortestPath((a:Person {name: $start})-[:KNOWS*]->(b:Person {name: $end})) "
"RETURN path",
start=start, end=end
)
return list(result)
# Usage
conn = Neo4jConnection("bolt://localhost:7687", "neo4j", "password")
conn.create_graph()
friends = conn.find_friends("Alice")
print("Alice's friends:", friends)
path = conn.shortest_path("Alice", "Charlie")
print("Shortest path:", path)
conn.close()📊 Big Data Overview
🎯 The 5 Vs of Big Data
Big Data is characterized by Volume, Velocity, Variety, Veracity, and Value. It requires specialized tools and approaches beyond traditional databases.
Big Data Characteristics
Volume: Massive amounts of data (terabytes to petabytes). Velocity: High-speed data generation and processing. Variety: Structured, semi-structured, unstructured data. Veracity: Data quality and trustworthiness. Value: Extracting insights from data.
Ecosystem: Hadoop, Spark, Kafka, Flink, NoSQL databases.
# Big data characteristics
class BigDataCharacteristics:
def __init__(self):
self.characteristics = {
'volume': 'Massive scale (petabytes+)',
'velocity': 'High speed (real-time)',
'variety': 'Multiple data types',
'veracity': 'Data quality challenges',
'value': 'Business insights'
}
def analyze_volume(self, data_size_gb):
if data_size_gb < 100:
return "Small data"
elif data_size_gb < 1000:
return "Big data (TB scale)"
else:
return "Very big data (PB+ scale)"
def analyze_velocity(self, records_per_second):
if records_per_second < 100:
return "Batch processing"
elif records_per_second < 10000:
return "Micro-batch"
else:
return "Real-time streaming"
# Distributed data processing
def distributed_processing(data, partitions=10):
# Split data into partitions for parallel processing
import multiprocessing as mp
def process_partition(partition_data):
# Process partition data
return partition_data.sum()
# Create partitions
partitions = np.array_split(data, partitions)
# Process in parallel
with mp.Pool(processes=partitions) as pool:
results = pool.map(process_partition, partitions)
return sum(results)🐘 Hadoop Ecosystem
🎯 The Hadoop Framework
Hadoop is a distributed computing framework for processing large datasets across clusters. It includes HDFS, MapReduce, YARN, and many companion tools.
Hadoop Components
HDFS: Distributed file system. MapReduce: Batch processing framework. YARN: Resource manager. Hive: SQL on Hadoop. HBase: Column-family DB. Pig: Data flow language.
Use cases: Batch ETL, data warehousing, log processing.
# Hadoop MapReduce (simplified)
from mrjob.job import MRJob
class WordCount(MRJob):
def mapper(self, _, line):
for word in line.split():
yield word, 1
def reducer(self, word, counts):
yield word, sum(counts)
# Hive SQL (conceptual)
hive_query = """
SELECT product, COUNT(*) as sales, SUM(amount) as revenue
FROM sales_data
WHERE sale_date >= '2024-01-01'
GROUP BY product
ORDER BY revenue DESC
LIMIT 10
"""
# HBase operations (Python)
import happybase
connection = happybase.Connection('localhost')
table = connection.table('users')
# Insert data
table.put(b'user_1001', {
b'info:name': b'Alice',
b'info:age': b'30',
b'info:city': b'NYC'
})
# Scan table
for key, data in table.scan(row_prefix=b'user_'):
print(f"Key: {key}, Data: {data}")
# Get specific row
row = table.row(b'user_1001')
print(row.get(b'info:name'))⚡ Apache Spark
🎯 Fast, Unified Analytics
Apache Spark is a unified analytics engine for large-scale data processing. It provides fast, in-memory processing and supports batch, streaming, SQL, ML, and graph processing.
Spark Features
In-memory: Caches data in memory for faster processing. Unified: One engine for all workloads. APIs: Python, Scala, Java, R. Libraries: Spark SQL, MLlib, Structured Streaming, GraphX.
Use cases: ETL, analytics, machine learning, real-time streams.
# Apache Spark (PySpark)
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import LinearRegression
# Create Spark session
spark = SparkSession.builder \
.appName("SparkDemo") \
.config("spark.sql.adaptive.enabled", "true") \
.getOrCreate()
# Create DataFrame
data = [("Alice", 25), ("Bob", 30), ("Charlie", 35)]
df = spark.createDataFrame(data, ["name", "age"])
# Show data
df.show()
# Read from CSV
df = spark.read.option("header", "true").csv("data.csv")
# Transformations
result = df \
.filter(col("age") > 25) \
.groupBy("department") \
.agg(avg("salary").alias("avg_salary")) \
.orderBy(col("avg_salary").desc())
# SQL queries
df.createOrReplaceTempView("employees")
result = spark.sql("""
SELECT department, AVG(salary) as avg_salary
FROM employees
WHERE age > 25
GROUP BY department
ORDER BY avg_salary DESC
""")
# Machine Learning
assembler = VectorAssembler(inputCols=["feature1", "feature2"], outputCol="features")
lr = LinearRegression(featuresCol="features", labelCol="target")
pipeline = Pipeline(stages=[assembler, lr])
model = pipeline.fit(train_data)
# Streaming
streaming_df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "localhost:9092") \
.option("subscribe", "events") \
.load()
streaming_df.selectExpr("CAST(value AS STRING)").writeStream \
.outputMode("append") \
.format("console") \
.start()🌊 Stream Processing
🎯 Real-Time Data Processing
Stream processing handles continuous data streams in real-time. It enables real-time analytics, monitoring, and event-driven applications.
Streaming Technologies
Apache Kafka: Distributed messaging. Apache Flink: Stateful streaming. Spark Streaming: Micro-batch processing. Apache Storm: Low-latency streaming.
Use cases: Real-time dashboards, fraud detection, IoT processing.
# Apache Kafka (Python)
from kafka import KafkaProducer, KafkaConsumer
import json
# Producer
producer = KafkaProducer(
bootstrap_servers=['localhost:9092'],
value_serializer=lambda v: json.dumps(v).encode('utf-8')
)
producer.send('events', {'user': 'Alice', 'action': 'login', 'timestamp': time.time()})
producer.send('events', {'user': 'Bob', 'action': 'purchase', 'amount': 99.99})
producer.flush()
# Consumer
consumer = KafkaConsumer(
'events',
bootstrap_servers=['localhost:9092'],
value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)
for message in consumer:
event = message.value
print(f"Received: {event}")
# Apache Flink (conceptual)
flink_query = """
SELECT
window_start,
event_type,
COUNT(*) as count,
SUM(value) as total
FROM events
GROUP BY window_start, event_type
ORDER BY window_start DESC
"""
# Kafka + Spark Streaming
streaming_data = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "localhost:9092") \
.option("subscribe", "events") \
.load()
# Windowed aggregation
windowed = streaming_data \
.select(from_json(col("value").cast("string"), schema).alias("data")) \
.select("data.*") \
.withColumn("timestamp", col("timestamp").cast("timestamp")) \
.groupBy(window("timestamp", "5 minutes"), "event_type") \
.agg(count("*").alias("count"))
# Write to console
windowed.writeStream \
.outputMode("update") \
.format("console") \
.start()📋 NoSQL vs RDBMS
🎯 Choosing the Right Database
Understanding the differences between NoSQL and RDBMS is crucial for choosing the right database for your application.
Comparison
RDBMS: Structured, ACID, complex joins, vertical scaling. NoSQL: Schema-less, BASE, horizontal scaling, specialized models. When to use: RDBMS for transactional apps; NoSQL for big data, real-time, flexible schemas.
Hybrid: Many applications use both (polyglot persistence).
# Comparison matrix
class DatabaseComparison:
def __init__(self):
self.comparison = {
'schema': {'rdbms': 'Fixed', 'nosql': 'Flexible'},
'scaling': {'rdbms': 'Vertical', 'nosql': 'Horizontal'},
'transactions': {'rdbms': 'ACID', 'nosql': 'BASE'},
'joins': {'rdbms': 'Complex', 'nosql': 'Limited'},
'use_case': {'rdbms': 'Transactional', 'nosql': 'Big Data'}
}
def recommend(self, requirements):
recommendations = []
if requirements.get('flexible_schema'):
recommendations.append('NoSQL')
if requirements.get('transactions'):
recommendations.append('RDBMS')
if requirements.get('scalability') == 'high':
recommendations.append('NoSQL')
if requirements.get('complex_queries'):
recommendations.append('RDBMS')
return recommendations
# Polyglot persistence example
class PolyglotPersistence:
def __init__(self):
self.rdbms = PostgreSQLConnection()
self.document = MongoDBConnection()
self.cache = RedisConnection()
def store_user_profile(self, user):
# Store in document DB (flexible schema)
self.document.insert('users', user)
# Store session in cache (fast access)
self.cache.set(f"user:{user['id']}", user)
# Store critical data in RDBMS (ACID)
self.rdbms.execute(
"INSERT INTO users (id, name, email) VALUES (%s, %s, %s)",
(user['id'], user['name'], user['email'])
)🔬 CAP Theorem
🎯 Consistency, Availability, Partition Tolerance
The CAP theorem states that a distributed system can only provide two of three guarantees: Consistency, Availability, and Partition Tolerance.
CAP in Practice
Consistency: All nodes see the same data. Availability: Every request gets a response. Partition Tolerance: System continues despite network partitions. Choosing: NoSQL databases make trade-offs (e.g., Cassandra: AP, MongoDB: CP).
Important: All systems must choose P; they choose between C and A.
# CAP theorem implementation
class CAPSystem:
def __init__(self, nodes):
self.nodes = nodes
self.type = 'CP' # or 'AP'
def write(self, key, value):
if self.type == 'CP':
# Strong consistency - wait for all nodes
responses = []
for node in self.nodes:
responses.append(node.write(key, value))
# Ensure all nodes wrote successfully
if all(responses):
return True
return False
else:
# AP - eventual consistency
for node in self.nodes:
node.write_async(key, value)
return True
def read(self, key):
if self.type == 'CP':
# Read from quorum
responses = []
for node in self.nodes:
responses.append(node.read(key))
# Return the latest (consistent) value
return self.resolve_conflicts(responses)
else:
# Read from any node
return self.nodes[0].read(key)
# Database CAP classifications
CAP_CLASSIFICATIONS = {
'Cassandra': 'AP (Eventual Consistency)',
'MongoDB': 'CP (Strong Consistency)',
'Redis': 'CP (Primary/Replica)',
'DynamoDB': 'AP (Eventual)',
'Neo4j': 'CP (ACID)',
'HBase': 'CP (Strong)'
}
# PACELC theorem (extension of CAP)
class PACELC(PAC):
def __init__(self):
self.choices = {
'Cassandra': 'AP/EL (Eventual Latency)',
'MongoDB': 'CP/EL (Strong Latency)',
'DynamoDB': 'AP/EC (Eventual Consistency)'
}💼 Applications
🎯 Real-World Use Cases
NoSQL and Big Data technologies power web-scale applications, IoT, real-time analytics, and machine learning across industries.
Application Domains
E-commerce: Product catalogs, recommendations, shopping carts. IoT: Sensor data processing, device management. Social Media: User profiles, feeds, relationships. Financial: Fraud detection, real-time trading.
Examples: Uber (Cassandra), Netflix (Cassandra/S3), Google (BigTable).
# Application: E-commerce recommendation
class RecommendationEngine:
def __init__(self, graph_db, document_db):
self.graph = graph_db
self.documents = document_db
def get_user_recommendations(self, user_id):
# Query graph for purchase patterns
similar_users = self.graph.query(
"MATCH (u:User {id: $id})-[:BOUGHT]->(p:Product) "
"MATCH (u2:User)-[:BOUGHT]->(p) "
"WHERE u2.id != $id "
"RETURN u2.id, COUNT(*) as common "
"ORDER BY common DESC LIMIT 10",
id=user_id
)
# Get products from documents
recommendations = []
for user in similar_users:
products = self.documents.find(
{'user_id': user['id']},
{'product_id': 1, '_id': 0}
)
recommendations.extend(products)
return recommendations
# Application: IoT sensor processing
class IoTPlatform:
def __init__(self, kafka, cassandra, spark):
self.kafka = kafka
self.cassandra = cassandra
self.spark = spark
def process_sensor_data(self):
# Stream data from Kafka
stream = self.spark.readStream.format("kafka").load()
# Process stream
processed = stream.selectExpr("CAST(value AS STRING) as json") \
.select(from_json(col("json"), schema).alias("data")) \
.select("data.*")
# Write to Cassandra
processed.writeStream \
.foreachBatch(lambda df, epoch: df.write \
.format("org.apache.spark.sql.cassandra") \
.options(table="sensor_data", keyspace="iot") \
.mode("append") \
.save()) \
.start()
# Application: Social network analytics
class SocialAnalytics:
def __init__(self, graph_db, redis):
self.graph = graph_db
self.cache = redis
def get_friend_suggestions(self, user_id):
# Check cache first
cached = self.cache.get(f"suggestions:{user_id}")
if cached:
return json.loads(cached)
# Query graph
suggestions = self.graph.query(
"MATCH (u:User {id: $id})-[:KNOWS*2]-(fof) "
"WHERE NOT (u)-[:KNOWS]-(fof) AND fof.id != $id "
"RETURN fof.id, COUNT(*) as score "
"ORDER BY score DESC LIMIT 10",
id=user_id
)
# Cache results
self.cache.setex(f"suggestions:{user_id}", 3600, json.dumps(suggestions))
return suggestions🚀 Future Directions
🎯 The Future of NoSQL and Big Data
NoSQL and Big Data continue to evolve with AI integration, serverless databases, and real-time analytics.
Emerging Trends
AI-Native: Databases with built-in ML capabilities. Serverless: Auto-scaling, pay-per-query. Edge: Distributed databases at the edge. Real-Time: Sub-second analytics. Multi-Model: Databases supporting multiple data models.
Unified: Single platform for all data needs.
# Future: AI-Native Database
class AINativeDB:
def __init__(self):
self.model = self.load_model()
self.data = {}
def insert(self, key, value):
# Auto-index with ML
embeddings = self.model.encode(value)
self.data[key] = {'value': value, 'embedding': embeddings}
def semantic_search(self, query):
# Search by meaning, not just keywords
query_embedding = self.model.encode(query)
results = []
for key, data in self.data.items():
similarity = cosine_similarity(query_embedding, data['embedding'])
results.append((key, similarity))
return sorted(results, key=lambda x: x[1], reverse=True)
# Future: Serverless NoSQL
class ServerlessDatabase:
def __init__(self, provider='aws'):
self.provider = provider
self.table = None
def create_table(self, name, config):
# Auto-scaling table
self.table = self.provider.create_table(name, config)
self.table.auto_scale(min=1, max=100)
return self.table
def query(self, query):
# Pay per query
result = self.table.query(query)
self.bill_for_operation(result.compute_units)
return result
# Future: Multi-Model Database
class MultiModelDB:
def __init__(self):
self.models = {
'document': DocumentStore(),
'graph': GraphDatabase(),
'keyvalue': KeyValueStore(),
'timeseries': TimeSeriesStore()
}
def use_model(self, model_type):
if model_type in self.models:
return self.models[model_type]
raise ValueError(f"Unknown model: {model_type}")
# Future: Edge Database
class EdgeDB:
def __init__(self, location):
self.location = location
self.local_data = {}
self.sync_with_cloud()
def sync_with_cloud(self):
# Sync local data with cloud
pass
def query_local(self, query):
# Query local data first
if self.local_data:
return self.local_data.query(query)
# Fall back to cloud
return self.cloud.query(query)