MongoDB Time Series Collections Kullanımı

Zaman serisi verisi, modern altyapıların en kritik sorunlarından biri haline geldi. IoT cihazları, uygulama metrikleri, finansal işlemler, sunucu logları… Bunların hepsi zamana bağlı, sıralı ve çok büyük hacimli veriler üretiyor. MongoDB 5.0 ile gelen Time Series Collections özelliği, bu tür verileri saklamak ve sorgulamak için özel olarak optimize edilmiş bir yapı sunuyor. Klasik koleksiyonlara kıyasla hem depolama alanında hem de sorgu performansında ciddi kazanımlar elde edebiliyorsunuz. Bu yazıda gerçek dünya senaryolarıyla MongoDB zaman serisi koleksiyonlarını derinlemesine inceleyeceğiz.

Zaman Serisi Koleksiyonu Nedir ve Neden Önemli?

Klasik bir MongoDB koleksiyonunda zaman damgalı veri tutmak mümkün, ama verimli değil. Her belge ayrı ayrı saklanır, sıkıştırma optimizasyonu yoktur ve büyük hacimlerde sorgular yavaşlar. Örneğin 100 sensörden saniyede bir ölçüm aldığınızı düşünün. Günde 8.64 milyon belge oluşuyor. Klasik koleksiyonla bunu yönetmek giderek zorlaşır.

Time Series Collections bu sorunu çözmek için şu yaklaşımı kullanır: Aynı kaynaktan (aynı sensör, aynı sunucu, aynı hisse senedi) gelen verileri zaman pencerelerine göre gruplar ve dahili olarak optimize edilmiş bir formatta saklar. Bu sayede:

Depolama tasarrufu: Veriler dahili olarak sıkıştırılır, klasik koleksiyona göre %50-90 daha az yer kaplar
Hızlı okuma: Zaman aralığı sorguları için özel indeks yapısı kullanılır
Otomatik TTL: Eski verileri otomatik silmek için yerleşik destek gelir
Bucket optimizasyonu: MongoDB, aynı kaynaktan gelen verileri “bucket” adı verilen dahili belgeler halinde gruplar

İlk Time Series Koleksiyonunu Oluşturma

Hemen pratikle başlayalım. Bir sunucu izleme sistemi için CPU ve bellek metriklerini saklayacağız.

# MongoDB shell'e bağlan
mongosh "mongodb://localhost:27017/monitoring"

# Time series koleksiyonu oluştur
db.createCollection("server_metrics", {
  timeseries: {
    timeField: "timestamp",      # Zorunlu: zaman damgası alanı
    metaField: "server_info",    # Opsiyonel: kaynak tanımlayıcı
    granularity: "seconds"       # seconds | minutes | hours
  },
  expireAfterSeconds: 2592000    # 30 gün sonra otomatik sil
})

Parametreleri açıklayalım:

timeField: Her belgede zaman damgasını tutan alan. Date tipi olmalı, zorunlu
metaField: Verinin kaynağını tanımlayan alan (sunucu adı, sensör ID’si vs.). Bu alan üzerinde gruplama yapılır
granularity: Verinin ne sıklıkla geldiğini MongoDB’ye söyler. Bucket boyutunu optimize eder
expireAfterSeconds: TTL süresi, eski verilerin otomatik temizlenmesi için

Şimdi bazı örnek veri ekleyelim:

# Tek belge ekleme
db.server_metrics.insertOne({
  timestamp: new Date(),
  server_info: {
    hostname: "web-01.prod.example.com",
    datacenter: "istanbul",
    environment: "production"
  },
  cpu_percent: 67.3,
  memory_percent: 82.1,
  disk_io_read: 1024,
  disk_io_write: 512,
  network_in: 2048,
  network_out: 1536
})

# Toplu veri ekleme (gerçekçi senaryo)
const metrics = []
const servers = ["web-01", "web-02", "db-01", "cache-01"]
const now = new Date()

for (let i = 0; i < 1440; i++) {  // 24 saatlik veri, dakika başı
  servers.forEach(server => {
    metrics.push({
      timestamp: new Date(now.getTime() - (i * 60000)),
      server_info: {
        hostname: `${server}.prod.example.com`,
        datacenter: "istanbul"
      },
      cpu_percent: Math.random() * 100,
      memory_percent: 60 + Math.random() * 35,
      disk_io_read: Math.floor(Math.random() * 5000),
      disk_io_write: Math.floor(Math.random() * 2000)
    })
  })
}

db.server_metrics.insertMany(metrics)

Granularity Ayarını Doğru Yapmak

Bu konu pratikte en çok gözden kaçan detaylardan biri. Granularity yanlış ayarlanırsa performans kazanımları yarıya düşebilir.

seconds: Saniyede birden fazla veri geliyorsa veya saniye bazında ölçüm yapılıyorsa
minutes: Dakika başı ölçüm için (örneğin sistem metrikleri)
hours: Saat başı veya daha seyrek ölçüm için (örneğin günlük raporlar)

Granularity, MongoDB’nin dahili bucket boyutunu belirler. “seconds” seçilirse her bucket yaklaşık 1 saatlik veri içerir, “minutes” seçilirse 24 saatlik, “hours” seçilirse 30 günlük. Yanlış seçim, bucket başına çok az veri düşmesi anlamına gelir ve sıkıştırma verimliliği düşer.

Mevcut bir koleksiyonun granularity’sini değiştiremezsiniz, bu yüzden baştan doğru seçmek önemli. Değiştirmeniz gerekirse yeni koleksiyon oluşturup veriyi migrate etmek zorundasınız.

Gerçek Dünya Senaryosu: IoT Sensör Verisi

Bir fabrika otomasyonu senaryosu düşünelim. 50 üretim hattında sıcaklık, basınç ve titreşim sensörleri var, saniyede veri gönderiyor.

# IoT sensör koleksiyonu
db.createCollection("factory_sensors", {
  timeseries: {
    timeField: "ts",
    metaField: "sensor",
    granularity: "seconds"
  },
  expireAfterSeconds: 7776000  # 90 günlük veri sakla
})

# İndeks ekle: sensör bazında sorgular için
db.factory_sensors.createIndex({
  "sensor.line_id": 1,
  "sensor.sensor_type": 1,
  "ts": 1
})

# Veri ekleme
db.factory_sensors.insertMany([
  {
    ts: new Date("2024-01-15T10:00:00Z"),
    sensor: {
      id: "TEMP-001",
      line_id: "LINE-A",
      sensor_type: "temperature",
      location: "station_3"
    },
    value: 78.5,
    unit: "celsius",
    status: "normal"
  },
  {
    ts: new Date("2024-01-15T10:00:00Z"),
    sensor: {
      id: "PRESS-001",
      line_id: "LINE-A",
      sensor_type: "pressure",
      location: "station_3"
    },
    value: 4.2,
    unit: "bar",
    status: "normal"
  }
])

Zaman Serisi Sorguları

Time Series Collections, standart MongoDB sorgu sözdizimini destekler ama bazı aggregation pipeline aşamaları özellikle optimize edilmiştir.

# Son 1 saatteki verileri getir
db.factory_sensors.find({
  "sensor.line_id": "LINE-A",
  "sensor.sensor_type": "temperature",
  ts: {
    $gte: new Date(Date.now() - 3600000),
    $lte: new Date()
  }
}).sort({ ts: 1 })

# $setWindowFields ile hareketli ortalama hesapla
db.server_metrics.aggregate([
  {
    $match: {
      "server_info.hostname": "web-01.prod.example.com",
      timestamp: {
        $gte: new Date(Date.now() - 86400000)  // Son 24 saat
      }
    }
  },
  {
    $setWindowFields: {
      partitionBy: "$server_info.hostname",
      sortBy: { timestamp: 1 },
      output: {
        cpu_moving_avg: {
          $avg: "$cpu_percent",
          window: { documents: [-5, 0] }  // Son 5 kayıt ortalaması
        }
      }
    }
  },
  {
    $project: {
      timestamp: 1,
      cpu_percent: 1,
      cpu_moving_avg: { $round: ["$cpu_moving_avg", 2] }
    }
  }
])

$densify ile Eksik Zaman Noktalarını Doldurmak

Gerçek dünyada sensörler bazen veri gönderemez, ağ kesintisi olur, cihaz yeniden başlar. Bu durumda zaman serinizde boşluklar oluşur. $densify aşaması bu boşlukları doldurur.

db.factory_sensors.aggregate([
  {
    $match: {
      "sensor.id": "TEMP-001",
      ts: {
        $gte: new Date("2024-01-15T08:00:00Z"),
        $lte: new Date("2024-01-15T12:00:00Z")
      }
    }
  },
  {
    $densify: {
      field: "ts",
      partitionByFields: ["sensor.id"],
      range: {
        step: 1,
        unit: "minute",
        bounds: "partition"
      }
    }
  },
  {
    $fill: {
      partitionByFields: ["sensor.id"],
      sortBy: { ts: 1 },
      output: {
        value: { method: "linear" }  # Lineer interpolasyon
      }
    }
  }
])

$bucket ve $dateTrunc ile Zaman Bazlı Gruplama

Zaman serisi verilerinde en sık yapılan işlem belirli zaman pencerelerine göre istatistik hesaplamak. Örneğin “son 24 saatte saatlik ortalama CPU kullanımı”.

# Saatlik ortalamalar
db.server_metrics.aggregate([
  {
    $match: {
      "server_info.hostname": "web-01.prod.example.com",
      timestamp: {
        $gte: new Date(Date.now() - 86400000)
      }
    }
  },
  {
    $group: {
      _id: {
        hour: {
          $dateTrunc: {
            date: "$timestamp",
            unit: "hour"
          }
        },
        hostname: "$server_info.hostname"
      },
      avg_cpu: { $avg: "$cpu_percent" },
      max_cpu: { $max: "$cpu_percent" },
      min_cpu: { $min: "$cpu_percent" },
      avg_memory: { $avg: "$memory_percent" },
      sample_count: { $sum: 1 }
    }
  },
  {
    $sort: { "_id.hour": 1 }
  },
  {
    $project: {
      hour: "$_id.hour",
      hostname: "$_id.hostname",
      avg_cpu: { $round: ["$avg_cpu", 2] },
      max_cpu: { $round: ["$max_cpu", 2] },
      min_cpu: { $round: ["$min_cpu", 2] },
      avg_memory: { $round: ["$avg_memory", 2] },
      sample_count: 1,
      _id: 0
    }
  }
])

Performans Analizi ve İndeks Stratejisi

Time Series Collections otomatik olarak timeField üzerinde bir indeks oluşturur. Ama ek sorgularınız varsa manuel indeks eklemek gerekir.

# Mevcut indeksleri görüntüle
db.server_metrics.getIndexes()

# Bileşik indeks: meta field + time
db.server_metrics.createIndex({
  "server_info.hostname": 1,
  "server_info.datacenter": 1,
  "timestamp": 1
})

# explain() ile sorgu planını incele
db.server_metrics.find({
  "server_info.hostname": "web-01.prod.example.com",
  timestamp: {
    $gte: new Date(Date.now() - 3600000)
  }
}).explain("executionStats")

# Koleksiyon istatistiklerini görüntüle
db.runCommand({
  collStats: "server_metrics",
  scale: 1048576  # MB cinsinden
})

İndeks stratejisinde dikkat edilecek noktalar:

metaField alt alanları üzerinde indeks oluşturmak filtreleme sorgularını ciddi ölçüde hızlandırır
Düşük kardinaliteli meta alanları için (örneğin datacenter: 3-4 değer) indeks pek fayda sağlamaz
Yüksek kardinaliteli meta alanları için (örneğin sensör ID’si: binlerce değer) indeks kritik önem taşır
Time field üzerinde zaten otomatik indeks var, tekrar oluşturmaya gerek yok

Python ile Time Series Koleksiyonu Yönetimi

Gerçek dünyada koleksiyonları elle yönetmiyorsunuz. İşte bir Python betiği:

# Gerekli kütüphaneyi kur
pip install pymongo

# Python scripti: factory_monitor.py
cat << 'EOF' > factory_monitor.py
from pymongo import MongoClient
from datetime import datetime, timedelta
import random
import time

client = MongoClient("mongodb://localhost:27017/")
db = client["factory_monitoring"]

# Koleksiyonu oluştur (yoksa)
if "sensor_data" not in db.list_collection_names():
    db.create_collection("sensor_data", {
        "timeseries": {
            "timeField": "timestamp",
            "metaField": "device",
            "granularity": "seconds"
        },
        "expireAfterSeconds": 604800  # 7 gün
    })
    print("Koleksiyon olusturuldu")

collection = db["sensor_data"]

def simulate_sensor_data(num_sensors=10, duration_minutes=5):
    """Sensör verisi simüle et ve kaydet"""
    sensors = [f"SENSOR-{i:03d}" for i in range(1, num_sensors + 1)]
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(minutes=duration_minutes)
    
    bulk_data = []
    current_time = start_time
    
    while current_time <= end_time:
        for sensor_id in sensors:
            bulk_data.append({
                "timestamp": current_time,
                "device": {
                    "id": sensor_id,
                    "type": "temperature",
                    "location": f"zone_{random.randint(1,5)}"
                },
                "temperature": round(20 + random.gauss(0, 5), 2),
                "humidity": round(50 + random.gauss(0, 10), 2),
                "battery_level": random.randint(20, 100)
            })
        current_time += timedelta(seconds=30)
    
    if bulk_data:
        result = collection.insert_many(bulk_data)
        print(f"{len(result.inserted_ids)} belge eklendi")

def get_hourly_stats(sensor_id, hours=24):
    """Saatlik istatistikleri getir"""
    pipeline = [
        {
            "$match": {
                "device.id": sensor_id,
                "timestamp": {
                    "$gte": datetime.utcnow() - timedelta(hours=hours)
                }
            }
        },
        {
            "$group": {
                "_id": {
                    "$dateTrunc": {"date": "$timestamp", "unit": "hour"}
                },
                "avg_temp": {"$avg": "$temperature"},
                "max_temp": {"$max": "$temperature"},
                "min_temp": {"$min": "$temperature"},
                "count": {"$sum": 1}
            }
        },
        {"$sort": {"_id": 1}}
    ]
    return list(collection.aggregate(pipeline))

# Veri simülasyonunu çalıştır
simulate_sensor_data(num_sensors=20, duration_minutes=60)

# İstatistikleri yazdır
stats = get_hourly_stats("SENSOR-001", hours=24)
for stat in stats:
    print(f"Saat: {stat['_id']} | Ort: {stat['avg_temp']:.1f}C | "
          f"Max: {stat['max_temp']:.1f}C | Kayit: {stat['count']}")

client.close()
EOF

python3 factory_monitor.py

TTL ve Veri Yaşam Döngüsü Yönetimi

Zaman serisi verilerinde veri yaşam döngüsü kritik bir konu. Her şeyi sonsuza kadar saklamak ne ekonomik ne de pratik.

# Mevcut TTL süresini görüntüle
db.runCommand({
  listCollections: 1,
  filter: { name: "server_metrics" }
})

# TTL süresini güncelle (collMod ile)
db.runCommand({
  collMod: "server_metrics",
  expireAfterSeconds: 5184000  # 60 gün
})

# Manuel temizleme: belirli tarihten öncekileri sil
# NOT: Time series'de doğrudan delete yapılabilir ama dikkatli olun
db.server_metrics.deleteMany({
  timestamp: {
    $lt: new Date(Date.now() - 7776000000)  // 90 günden eski
  }
})

# Koleksiyon boyutunu kontrol et
db.server_metrics.stats().then(stats => {
  print(`Toplam boyut: ${(stats.size / 1048576).toFixed(2)} MB`)
  print(`Belge sayisi: ${stats.count}`)
  print(`Ortalama belge boyutu: ${stats.avgObjSize} bytes`)
})

Veri saklama stratejisi için pratik bir yaklaşım: ham verileri kısa süre saklayın (7-30 gün), aggregation ile özetlenmiş verileri daha uzun süre ayrı bir koleksiyonda tutun. Örneğin dakika bazlı ham veriyi 30 gün saklarken saatlik özetleri 1 yıl, günlük özetleri 5 yıl saklayabilirsiniz.

Monitoring ve Yönetim Komutları

# Time series koleksiyonlarını listele
db.runCommand({
  listCollections: 1,
  filter: { type: "timeseries" }
})

# Koleksiyon detaylarını görüntüle
db.runCommand({
  listCollections: 1,
  filter: { name: "server_metrics" }
})

# Bucket istatistiklerini görüntüle (dahili sistem koleksiyonu)
db.system.buckets.server_metrics.stats()

# Aggregation pipeline performansını ölç
db.server_metrics.aggregate(
  [
    {
      $match: {
        timestamp: { $gte: new Date(Date.now() - 3600000) }
      }
    },
    {
      $group: {
        _id: "$server_info.hostname",
        avg_cpu: { $avg: "$cpu_percent" }
      }
    }
  ],
  { explain: true }
)

Sık Yapılan Hatalar ve Çözümleri

Sahada gördüğüm yaygın sorunları paylaşayım:

timeField’a yanlış tip veri göndermek: timeField mutlaka Date tipinde olmalı. String olarak "2024-01-15T10:00:00Z" gönderirseniz hata alırsınız. MongoDB driver’ınızda datetime objesi oluşturup gönderdiğinizden emin olun.

metaField olmadan koleksiyon oluşturmak: Teknik olarak mümkün, ama ciddi performans kaybına yol açar. Verileriniz birden fazla kaynaktan geliyorsa mutlaka metaField kullanın.

Çok derin iç içe metaField yapısı: sensor.location.building.floor.room gibi derin hiyerarşiler sorgu yazmayı ve indekslemeyi zorlaştırır. metaField’ı mümkün olduğunca düz tutun.

granularity’yi verisizlerin sıklığına göre ayarlamamak: Saatte bir veri gelen sensörler için granularity: "seconds" seçmek bucket optimizasyonunu bozar.

Update işlemlerini yanlış anlamak: Time Series Collections’da belgeler güncellenmez, sadece eklenir. Var olan bir belgeyi update etmeye çalışırsanız hata alırsınız. Bu yapı append-only (sadece ekleme) çalışır.

Sonuç

MongoDB Time Series Collections, zaman serisi verisi için özel olarak tasarlanmış ve klasik koleksiyonlara göre çok daha verimli bir yapı sunuyor. Özellikle IoT projelerinde, uygulama izleme sistemlerinde ve finansal veri uygulamalarında ciddi depolama tasarrufu ve sorgu hızı artışı elde edebilirsiniz.

Önemli noktaları özetleyelim:

granularity seçimini verinin gerçek frekansına göre yapın
metaField kullanmayı ihmal etmeyin, gruplama ve indeksleme için kritik
TTL ile veri yaşam döngüsünü baştan planlayın, disk dolduktan sonra temizlemek daha zor
$dateTrunc, $densify ve $setWindowFields aggregation aşamalarını öğrenin, zaman serisi analizlerini çok kolaylaştırıyor
Ham veriyi kısa süre, özetlenmiş veriyi uzun süre saklama stratejisini uygulayın

MongoDB 6.0 ve üzeri sürümlerde Time Series Collections’a eklenen yeni özellikler (partial TTL, daha esnek granularity vs.) bu yapıyı daha da güçlü kılıyor. Eğer hala zaman damgalı verileri klasik koleksiyonlarda tutuyorsanız, migration’ı ciddi olarak değerlendirmenizi öneririm. İlk migration biraz iş gerektiriyor ama uzun vadede hem maliyeti hem de yönetim yükünü önemli ölçüde azaltıyor.

MongoDB Time Series Collections ile Zaman Serisi Verisi Yönetimi