Part 1: Time Series Database Fundamentals and Architecture - Complete Guide to Modern TDB

# sensor_data_collector.py
import time
import random
import json
from datetime import datetime, timedelta
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import logging

class SensorDataCollector:
    def __init__(self):
        # InfluxDB client configuration
        self.client = InfluxDBClient(
            url="http://localhost:8086",
            token="admin-token",  # Replace with actual token
            org="myorg"
        )
        
        self.write_api = self.client.write_api(write_options=SYNCHRONOUS)
        self.query_api = self.client.query_api()
        
        # Logging configuration
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
    
    def generate_sensor_data(self, sensor_id, location, sensor_type):
        """Sensor data generation simulator"""
        
        # Data ranges by sensor type
        ranges = {
            'temperature': (15, 35),    # Temperature (Celsius)
            'humidity': (30, 80),       # Humidity (%)
            'pressure': (1000, 1030),   # Pressure (hPa)
            'vibration': (0, 10)        # Vibration (mm/s)
        }
        
        base_value = random.uniform(*ranges.get(sensor_type, (0, 100)))
        
        # Simulate time-based variation (sine wave)
        time_factor = time.time() / 3600  # Hour units
        variation = 2 * random.uniform(-1, 1) * abs(base_value * 0.1)
        
        value = base_value + variation
        
        return {
            'measurement': 'sensor_data',
            'tags': {
                'sensor_id': sensor_id,
                'location': location,
                'sensor_type': sensor_type,
                'status': 'active'
            },
            'fields': {
                'value': round(value, 2),
                'quality': random.randint(85, 100),
                'battery_level': random.randint(20, 100),
                'signal_strength': random.randint(-80, -30)
            },
            'timestamp': datetime.utcnow()
        }
    
    def write_sensor_data(self, data_points):
        """Store sensor data to InfluxDB"""
        try:
            points = []
            for data in data_points:
                point = Point(data['measurement']) \
                    .tag('sensor_id', data['tags']['sensor_id']) \
                    .tag('location', data['tags']['location']) \
                    .tag('sensor_type', data['tags']['sensor_type']) \
                    .tag('status', data['tags']['status']) \
                    .field('value', data['fields']['value']) \
                    .field('quality', data['fields']['quality']) \
                    .field('battery_level', data['fields']['battery_level']) \
                    .field('signal_strength', data['fields']['signal_strength']) \
                    .time(data['timestamp'])
                
                points.append(point)
            
            # Batch write
            self.write_api.write(bucket="mybucket", record=points)
            
            self.logger.info(f"✅ {len(points)} data points successfully stored.")
            return True
            
        except Exception as e:
            self.logger.error(f"❌ Data storage failed: {e}")
            return False
    
    def query_sensor_data(self, sensor_type=None, location=None, hours=1):
        """Query sensor data"""
        try:
            # Build Flux query
            query = f'''
            from(bucket: "mybucket")
            |> range(start: -{hours}h)
            |> filter(fn: (r) => r._measurement == "sensor_data")
            '''
            
            if sensor_type:
                query += f'|> filter(fn: (r) => r.sensor_type == "{sensor_type}")\n'
            
            if location:
                query += f'|> filter(fn: (r) => r.location == "{location}")\n'
            
            query += '|> sort(columns: ["_time"], desc: true)\n'
            query += '|> limit(n: 100)'
            
            # Execute query
            result = self.query_api.query(query)
            
            # Convert results
            data_points = []
            for table in result:
                for record in table.records:
                    data_points.append({
                        'time': record.get_time(),
                        'measurement': record.get_measurement(),
                        'field': record.get_field(),
                        'value': record.get_value(),
                        'sensor_id': record.values.get('sensor_id'),
                        'location': record.values.get('location'),
                        'sensor_type': record.values.get('sensor_type')
                    })
            
            return data_points
            
        except Exception as e:
            self.logger.error(f"❌ Data query failed: {e}")
            return []
    
    def get_statistics(self, sensor_type, hours=24):
        """Get sensor data statistics"""
        try:
            query = f'''
            from(bucket: "mybucket")
            |> range(start: -{hours}h)
            |> filter(fn: (r) => r._measurement == "sensor_data")
            |> filter(fn: (r) => r.sensor_type == "{sensor_type}")
            |> filter(fn: (r) => r._field == "value")
            |> group(columns: ["location"])
            |> mean()
            |> yield(name: "mean")
            
            from(bucket: "mybucket")
            |> range(start: -{hours}h)
            |> filter(fn: (r) => r._measurement == "sensor_data")
            |> filter(fn: (r) => r.sensor_type == "{sensor_type}")
            |> filter(fn: (r) => r._field == "value")
            |> group(columns: ["location"])
            |> max()
            |> yield(name: "max")
            
            from(bucket: "mybucket")
            |> range(start: -{hours}h)
            |> filter(fn: (r) => r._measurement == "sensor_data")
            |> filter(fn: (r) => r.sensor_type == "{sensor_type}")
            |> filter(fn: (r) => r._field == "value")
            |> group(columns: ["location"])
            |> min()
            |> yield(name: "min")
            '''
            
            result = self.query_api.query(query)
            
            stats = {}
            for table in result:
                table_name = table.name
                for record in table.records:
                    location = record.values.get('location')
                    if location not in stats:
                        stats[location] = {}
                    stats[location][table_name] = record.get_value()
            
            return stats
            
        except Exception as e:
            self.logger.error(f"❌ Statistics query failed: {e}")
            return {}
    
    def run_data_collection(self, duration_minutes=10):
        """Run data collection"""
        self.logger.info(f"🔄 Starting {duration_minutes}-minute sensor data collection...")
        
        # Sensor configuration
        sensors = [
            {'id': 'sensor_001', 'location': 'seoul', 'type': 'temperature'},
            {'id': 'sensor_002', 'location': 'seoul', 'type': 'humidity'},
            {'id': 'sensor_003', 'location': 'busan', 'type': 'temperature'},
            {'id': 'sensor_004', 'location': 'busan', 'type': 'pressure'},
            {'id': 'sensor_005', 'location': 'daegu', 'type': 'vibration'},
        ]
        
        start_time = time.time()
        end_time = start_time + (duration_minutes * 60)
        
        while time.time() < end_time:
            # Generate data from each sensor
            data_points = []
            for sensor in sensors:
                data = self.generate_sensor_data(
                    sensor['id'], 
                    sensor['location'], 
                    sensor['type']
                )
                data_points.append(data)
            
            # Store data
            self.write_sensor_data(data_points)
            
            # Wait 5 seconds
            time.sleep(5)
        
        self.logger.info("✅ Data collection completed!")
        
        # Print collection summary
        self.print_collection_summary()
    
    def print_collection_summary(self):
        """Print collection summary information"""
        print("\n📊 Data Collection Summary")
        print("=" * 50)
        
        sensor_types = ['temperature', 'humidity', 'pressure', 'vibration']
        
        for sensor_type in sensor_types:
            stats = self.get_statistics(sensor_type, hours=1)
            
            if stats:
                print(f"\n📈 {sensor_type.upper()} Sensor Statistics (Last 1 hour):")
                for location, values in stats.items():
                    mean_val = values.get('mean', 0)
                    max_val = values.get('max', 0)
                    min_val = values.get('min', 0)
                    print(f"   {location}: avg={mean_val:.2f}, max={max_val:.2f}, min={min_val:.2f}")

# Execution example
if __name__ == "__main__":
    collector = SensorDataCollector()
    
    # Run data collection (5 minutes)
    collector.run_data_collection(duration_minutes=5)
    
    # Query recent data
    print("\n🔍 Recent Temperature Sensor Data:")
    recent_data = collector.query_sensor_data(sensor_type='temperature', hours=1)
    
    for data in recent_data[:5]:  # Print only recent 5
        print(f"   {data['time']}: {data['location']} - {data['value']}")

# timescale_comparison.py
import psycopg2
import time
from datetime import datetime, timedelta
import random

class TimescaleComparison:
    def __init__(self):
        # TimescaleDB connection
        self.conn = psycopg2.connect(
            host="localhost",
            database="timeseries",
            user="postgres",
            password="postgres"
        )
        self.cur = self.conn.cursor()
        
        self.setup_timescale()
    
    def setup_timescale(self):
        """TimescaleDB initial setup"""
        # Enable TimescaleDB extension
        self.cur.execute("CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE;")
        
        # Create sensor data table
        self.cur.execute("""
            CREATE TABLE IF NOT EXISTS sensor_data (
                time TIMESTAMPTZ NOT NULL,
                sensor_id TEXT NOT NULL,
                location TEXT NOT NULL,
                sensor_type TEXT NOT NULL,
                value DOUBLE PRECISION NOT NULL,
                quality INTEGER NOT NULL,
                battery_level INTEGER NOT NULL,
                signal_strength INTEGER NOT NULL
            );
        """)
        
        # Convert to hypertable
        self.cur.execute("""
            SELECT create_hypertable('sensor_data', 'time', 
                                   if_not_exists => TRUE);
        """)
        
        # Create indexes
        self.cur.execute("""
            CREATE INDEX IF NOT EXISTS idx_sensor_data_sensor_type 
            ON sensor_data (sensor_type, time DESC);
        """)
        
        self.conn.commit()
        print("✅ TimescaleDB setup completed")
    
    def insert_data(self, num_records=1000):
        """Bulk data insertion test"""
        print(f"📝 Testing {num_records} record insertion...")
        
        start_time = time.time()
        
        # Batch insert
        insert_query = """
            INSERT INTO sensor_data 
            (time, sensor_id, location, sensor_type, value, quality, battery_level, signal_strength)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s);
        """
        
        data = []
        for i in range(num_records):
            timestamp = datetime.utcnow() - timedelta(seconds=i)
            data.append((
                timestamp,
                f'sensor_{i % 100:03d}',
                random.choice(['seoul', 'busan', 'daegu']),
                random.choice(['temperature', 'humidity', 'pressure']),
                random.uniform(20, 30),
                random.randint(80, 100),
                random.randint(20, 100),
                random.randint(-80, -30)
            ))
        
        self.cur.executemany(insert_query, data)
        self.conn.commit()
        
        end_time = time.time()
        duration = end_time - start_time
        
        print(f"✅ Insertion completed: {duration:.2f} seconds ({num_records/duration:.0f} records/sec)")
        return duration
    
    def query_performance_test(self):
        """Query performance test"""
        print("🔍 Query performance test...")
        
        queries = [
            {
                'name': 'Recent 1 hour data',
                'query': """
                    SELECT * FROM sensor_data 
                    WHERE time >= NOW() - INTERVAL '1 hour' 
                    ORDER BY time DESC 
                    LIMIT 100;
                """
            },
            {
                'name': 'Temperature sensor average',
                'query': """
                    SELECT location, AVG(value) as avg_temp
                    FROM sensor_data 
                    WHERE sensor_type = 'temperature' 
                    AND time >= NOW() - INTERVAL '1 hour'
                    GROUP BY location;
                """
            },
            {
                'name': 'Time window aggregation',
                'query': """
                    SELECT time_bucket('5 minutes', time) as bucket,
                           AVG(value) as avg_value
                    FROM sensor_data 
                    WHERE time >= NOW() - INTERVAL '1 hour'
                    GROUP BY bucket
                    ORDER BY bucket;
                """
            }
        ]
        
        results = {}
        
        for query_info in queries:
            start_time = time.time()
            
            self.cur.execute(query_info['query'])
            rows = self.cur.fetchall()
            
            end_time = time.time()
            duration = end_time - start_time
            
            results[query_info['name']] = {
                'duration': duration,
                'rows': len(rows)
            }
            
            print(f"   {query_info['name']}: {duration:.3f} seconds ({len(rows)} rows)")
        
        return results
    
    def compression_test(self):
        """Compression test"""
        print("🗜️ Compression test...")
        
        # Enable compression
        self.cur.execute("""
            ALTER TABLE sensor_data SET (
                timescaledb.compress,
                timescaledb.compress_segmentby = 'sensor_type, location'
            );
        """)
        
        # Set compression policy
        self.cur.execute("""
            SELECT add_compression_policy('sensor_data', INTERVAL '7 days');
        """)
        
        self.conn.commit()
        
        # Query compression statistics
        self.cur.execute("""
            SELECT 
                schemaname, tablename, 
                pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as size
            FROM pg_tables 
            WHERE tablename = 'sensor_data';
        """)
        
        size_info = self.cur.fetchone()
        print(f"   Table size: {size_info[2]}")
        
        return size_info
    
    def cleanup(self):
        """Cleanup"""
        self.cur.close()
        self.conn.close()

# Performance comparison execution
if __name__ == "__main__":
    print("🚀 TimescaleDB vs InfluxDB Performance Comparison")
    print("=" * 50)
    
    # TimescaleDB test
    tsdb = TimescaleComparison()
    
    # Data insertion performance
    insert_time = tsdb.insert_data(10000)
    
    # Query performance
    query_results = tsdb.query_performance_test()
    
    # Compression test
    compression_info = tsdb.compression_test()
    
    tsdb.cleanup()
    
    print("\n📊 Performance Comparison Results:")
    print(f"   TimescaleDB insertion: {insert_time:.2f} seconds")
    print("   Query performance:")
    for name, result in query_results.items():
        print(f"     {name}: {result['duration']:.3f} seconds")

# prometheus_metrics.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
import requests
import json
from datetime import datetime

# Prometheus metrics definition
sensor_data_total = Counter('sensor_data_points_total', 'Total sensor data points collected', ['sensor_type', 'location'])
data_quality_gauge = Gauge('sensor_data_quality', 'Sensor data quality score', ['sensor_type', 'location'])
battery_level_gauge = Gauge('sensor_battery_level', 'Sensor battery level', ['sensor_id'])
collection_duration = Histogram('data_collection_duration_seconds', 'Time spent collecting data')

class PrometheusMetricsCollector:
    def __init__(self, influxdb_url="http://localhost:8086"):
        self.influxdb_url = influxdb_url
        self.headers = {
            'Authorization': 'Token admin-token',
            'Content-Type': 'application/json'
        }
    
    @collection_duration.time()
    def collect_metrics(self):
        """Collect metrics from InfluxDB"""
        try:
            # Query data from last 5 minutes
            query = {
                'query': '''
                from(bucket: "mybucket")
                |> range(start: -5m)
                |> filter(fn: (r) => r._measurement == "sensor_data")
                |> group(columns: ["sensor_type", "location"])
                |> count()
                '''
            }
            
            response = requests.post(
                f"{self.influxdb_url}/api/v2/query",
                headers=self.headers,
                json=query
            )
            
            if response.status_code == 200:
                result = response.json()
                
                # Update metrics
                for table in result.get('results', [{}])[0].get('series', []):
                    for row in table.get('values', []):
                        sensor_type = row[0]  # sensor_type tag
                        location = row[1]     # location tag
                        count = row[2]        # count value
                        
                        sensor_data_total.labels(
                            sensor_type=sensor_type,
                            location=location
                        ).inc(count)
                
                # Quality score metrics
                self.collect_quality_metrics()
                
                # Battery level metrics
                self.collect_battery_metrics()
                
                print(f"✅ Metrics collection completed: {datetime.now()}")
                
            else:
                print(f"❌ Metrics collection failed: {response.status_code}")
                
        except Exception as e:
            print(f"❌ Metrics collection error: {e}")
    
    def collect_quality_metrics(self):
        """Collect quality score metrics"""
        query = {
            'query': '''
            from(bucket: "mybucket")
            |> range(start: -5m)
            |> filter(fn: (r) => r._measurement == "sensor_data")
            |> filter(fn: (r) => r._field == "quality")
            |> group(columns: ["sensor_type", "location"])
            |> mean()
            '''
        }
        
        try:
            response = requests.post(
                f"{self.influxdb_url}/api/v2/query",
                headers=self.headers,
                json=query
            )
            
            if response.status_code == 200:
                result = response.json()
                
                for table in result.get('results', [{}])[0].get('series', []):
                    for row in table.get('values', []):
                        sensor_type = row[0]
                        location = row[1]
                        quality = row[2]
                        
                        data_quality_gauge.labels(
                            sensor_type=sensor_type,
                            location=location
                        ).set(quality)
                        
        except Exception as e:
            print(f"Quality metrics collection error: {e}")
    
    def collect_battery_metrics(self):
        """Collect battery level metrics"""
        query = {
            'query': '''
            from(bucket: "mybucket")
            |> range(start: -5m)
            |> filter(fn: (r) => r._measurement == "sensor_data")
            |> filter(fn: (r) => r._field == "battery_level")
            |> group(columns: ["sensor_id"])
            |> last()
            '''
        }
        
        try:
            response = requests.post(
                f"{self.influxdb_url}/api/v2/query",
                headers=self.headers,
                json=query
            )
            
            if response.status_code == 200:
                result = response.json()
                
                for table in result.get('results', [{}])[0].get('series', []):
                    for row in table.get('values', []):
                        sensor_id = row[0]
                        battery_level = row[2]
                        
                        battery_level_gauge.labels(
                            sensor_id=sensor_id
                        ).set(battery_level)
                        
        except Exception as e:
            print(f"Battery metrics collection error: {e}")
    
    def run_metrics_server(self, port=8000):
        """Run metrics server"""
        print(f"🌐 Starting Prometheus metrics server: http://localhost:{port}/metrics")
        start_http_server(port)
        
        while True:
            self.collect_metrics()
            time.sleep(30)  # Collect every 30 seconds

if __name__ == "__main__":
    collector = PrometheusMetricsCollector()
    collector.run_metrics_server()