Elasticsearch Reindex API ile Veri Göçü

Tarafındankomut
0 Comments

Üretim ortamında bir Elasticsearch kümesini yönetirken er ya da geç şu soruyla yüz yüze gelirsin: “Bu index’i nasıl taşıyacağım, yeniden yapılandıracağım ya da dönüştüreceğim?” İşte tam bu noktada Reindex API devreye giriyor. Mapping değişiklikleri, shard sayısı güncellemeleri, cluster göçleri veya veri temizleme işlemleri için Reindex API, Elasticsearch’ün en güçlü araçlarından biri. Bu yazıda gerçek dünya senaryoları üzerinden Reindex API’yi derinlemesine inceleyeceğiz.

Reindex API Nedir ve Ne Zaman Kullanılır?

Elasticsearch’te bir index oluşturduktan sonra bazı şeyleri değiştiremezsin. Örneğin shard sayısını, field’ların mapping tipini veya analyzer ayarlarını doğrudan güncellemek mümkün değil. Bu tür değişiklikler için yeni bir index oluşturup veriyi oraya taşıman gerekiyor. Reindex API tam da bu işi yapıyor.

Reindex API’yi kullandığın başlıca durumlar:

  • Mapping değişikliği (örneğin bir field’ı text‘ten keyword‘e çevirmek)
  • Shard sayısını artırmak ya da azaltmak
  • Index ayarlarını güncellemek (replica sayısı, refresh interval vb.)
  • Elasticsearch versiyon yükseltmesi sırasında eski format index’leri taşımak
  • Farklı bir cluster’a veri göçü yapmak
  • Veriyi filtrelerek ya da dönüştürerek yeni bir index oluşturmak
  • Index isim şemasını değiştirmek

Temel Reindex Kullanımı

En basit haliyle Reindex, bir source index’ten bir destination index’e tüm belgeleri kopyalar.

curl -X POST "localhost:9200/_reindex" -H 'Content-Type: application/json' -d'
{
  "source": {
    "index": "urunler_v1"
  },
  "dest": {
    "index": "urunler_v2"
  }
}'

Bu komut urunler_v1 içindeki tüm belgeleri urunler_v2‘ye kopyalar. Destination index yoksa Elasticsearch otomatik olarak oluşturur, ancak bu durumda mapping’i source’dan miras alır. Eğer mapping değişikliği yapacaksan destination index’i önceden oluşturman gerekir.

Önemli bir nokta: Reindex işlemi source index’teki verileri silmez, sadece kopyalar. Bu yüzden işlem sonrası eski index’i manuel olarak silmen gerekir.

Destination Index’i Önceden Hazırlamak

Gerçek dünyada neredeyse her zaman destination index’i önceden hazırlarsın. Diyelim ki bir e-ticaret platformunun ürün index’ini yeniden yapılandırıyorsun ve fiyat field’ını float‘tan scaled_float‘a çevirmek istiyorsun.

# Önce yeni mapping ile destination index'i oluştur
curl -X PUT "localhost:9200/urunler_v2" -H 'Content-Type: application/json' -d'
{
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1,
    "refresh_interval": "-1"
  },
  "mappings": {
    "properties": {
      "urun_adi": {
        "type": "text",
        "analyzer": "turkish"
      },
      "fiyat": {
        "type": "scaled_float",
        "scaling_factor": 100
      },
      "kategori": {
        "type": "keyword"
      },
      "olusturma_tarihi": {
        "type": "date",
        "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"
      }
    }
  }
}'

Burada refresh_interval: -1 ayarına dikkat et. Reindex sırasında otomatik refresh’i devre dışı bırakmak işlemi ciddi ölçüde hızlandırır. İşlem bittikten sonra bu değeri normale döndürmeyi unutma.

Filtrelenmiş Reindex

Sadece belirli kriterlere uyan belgeleri taşımak istediğinde query ekleyebilirsin. Örneğin sadece aktif ürünleri yeni index’e taşımak:

curl -X POST "localhost:9200/_reindex" -H 'Content-Type: application/json' -d'
{
  "source": {
    "index": "urunler_v1",
    "query": {
      "bool": {
        "must": [
          {"term": {"durum": "aktif"}},
          {"range": {
            "stok_miktari": {
              "gt": 0
            }
          }}
        ]
      }
    },
    "size": 1000
  },
  "dest": {
    "index": "urunler_aktif_v2"
  }
}'

size parametresi her scroll batch’inde kaç belge alınacağını belirler. Varsayılan değer 1000’dir. Büyük belgeler için bu değeri düşürmek, küçük belgeler için artırmak performansı etkiler.

Script ile Veri Dönüşümü

Reindex’in en güçlü özelliklerinden biri, taşıma sırasında veriyi dönüştürebilmesidir. Painless script kullanarak field ekleyebilir, silebilir veya değerlerini değiştirebilirsin.

Diyelim ki eski index’te ad ve soyad ayrı field’larda tutulmuş, yeni index’te bunları birleştirip tam_ad field’ı oluşturmak istiyorsun:

curl -X POST "localhost:9200/_reindex" -H 'Content-Type: application/json' -d'
{
  "source": {
    "index": "musteriler_v1"
  },
  "dest": {
    "index": "musteriler_v2"
  },
  "script": {
    "lang": "painless",
    "source": """
      ctx._source.tam_ad = ctx._source.ad + " " + ctx._source.soyad;
      ctx._source.remove("ad");
      ctx._source.remove("soyad");
      
      if (ctx._source.kayit_tarihi != null) {
        ctx._source.kayit_yili = ctx._source.kayit_tarihi.substring(0, 4);
      }
      
      if (ctx._source.toplam_siparis == null) {
        ctx._source.toplam_siparis = 0;
      }
    """
  }
}'

Bu script’te birkaç işlem yapıyoruz: ad ve soyad‘ı birleştirip tam_ad oluşturuyoruz, eski field’ları siliyoruz, kayıt tarihinden yıl bilgisini ayırıyoruz ve null kontrolleri yapıyoruz.

Asenkron Reindex ve Task API

Büyük index’lerde Reindex işlemi saatler sürebilir. Bu durumda işlemi asenkron olarak başlatmak ve takip etmek gerekir.

# wait_for_completion=false ile asenkron başlat
curl -X POST "localhost:9200/_reindex?wait_for_completion=false" 
  -H 'Content-Type: application/json' -d'
{
  "source": {
    "index": "buyuk_log_index_v1"
  },
  "dest": {
    "index": "buyuk_log_index_v2"
  }
}'

# Yanıt olarak task ID alırsın:
# {"task": "r1A2WoRbTwKZ516z6NEs5A:36619"}

Bu komut çalıştırıldığında Elasticsearch bir task ID döndürür. Bu ID ile işlemi takip edebilirsin:

# Task durumunu kontrol et
curl -X GET "localhost:9200/_tasks/r1A2WoRbTwKZ516z6NEs5A:36619"

# Tüm aktif reindex task'larını listele
curl -X GET "localhost:9200/_tasks?actions=*reindex&detailed=true"

# İşlemi iptal etmek gerekirse
curl -X POST "localhost:9200/_tasks/r1A2WoRbTwKZ516z6NEs5A:36619/_cancel"

Task durumu sorguladığında şu bilgileri görebilirsin:

  • total: Toplam taşınacak belge sayısı
  • created: Oluşturulan belge sayısı
  • updated: Güncellenen belge sayısı
  • deleted: Silinen belge sayısı
  • version_conflicts: Versiyon çakışmaları
  • status.throttled_millis: Throttling nedeniyle geçen süre

Throttling ile Üretim Ortamında Güvenli Reindex

Üretim ortamında Reindex yaparken cluster’ı bunaltmamak kritik önem taşır. requests_per_second parametresi ile throttling uygulayabilirsin:

curl -X POST "localhost:9200/_reindex?wait_for_completion=false&requests_per_second=500" 
  -H 'Content-Type: application/json' -d'
{
  "source": {
    "index": "canli_urunler_v1",
    "size": 500
  },
  "dest": {
    "index": "canli_urunler_v2"
  },
  "conflicts": "proceed"
}'

conflicts: proceed parametresi önemli: Varsayılan olarak bir versiyon çakışması işlemi durdurur. proceed diyerek çakışmaları atla ve devam et demiş oluyorsun. Canlı bir index’te okuma-yazma devam ederken reindex yapıyorsan bu parametreyi kullanman şart.

Çalışan bir task’ın throttle değerini dinamik olarak değiştirebilirsin:

# Throttle değerini artır (yoğun saatlerde düşür, gece düşük trafikte artır)
curl -X POST "localhost:9200/_reindex/r1A2WoRbTwKZ516z6NEs5A:36619/_rethrottle?requests_per_second=1000"

# Throttle'ı tamamen kaldır
curl -X POST "localhost:9200/_reindex/r1A2WoRbTwKZ516z6NEs5A:36619/_rethrottle?requests_per_second=-1"

Farklı Cluster’a Veri Göçü

Elasticsearch’ün gerçek gücü, farklı bir cluster’daki index’ten veri çekebilmesinde yatıyor. Bu özelliği kullanmak için önce elasticsearch.yml dosyasında remote cluster’ı whitelist’e eklemelisin.

# /etc/elasticsearch/elasticsearch.yml dosyasına ekle
# reindex.remote.whitelist: "eski-cluster.sirket.com:9200"

# Ardından elasticsearch servisini yeniden başlat
sudo systemctl restart elasticsearch

# Remote cluster'dan reindex
curl -X POST "localhost:9200/_reindex" -H 'Content-Type: application/json' -d'
{
  "source": {
    "remote": {
      "host": "http://eski-cluster.sirket.com:9200",
      "username": "elastic",
      "password": "gizli_sifre",
      "socket_timeout": "1m",
      "connect_timeout": "10s"
    },
    "index": "urunler",
    "query": {
      "match_all": {}
    },
    "size": 500
  },
  "dest": {
    "index": "urunler_v2"
  }
}'

Remote reindex’te size parametresini daha düşük tutmak genellikle daha güvenlidir çünkü ağ gecikmesi ve bant genişliği sınırlamaları devreye girer. Büyük veri setleri için 100-500 arası bir değer mantıklı bir başlangıç noktası.

Alias ile Sıfır Kesinti Süreli Geçiş

Gerçek üretim ortamlarında Reindex’in en kritik kısmı, uygulamanın kesintisiz çalışmaya devam etmesi. Bunu alias’larla sağlarsın. İşte tipik bir canlı geçiş akışı:

# 1. Mevcut durumu kontrol et - uygulama "urunler" alias'ını kullanıyor
curl -X GET "localhost:9200/_alias/urunler"

# 2. Yeni index'i hazırla
curl -X PUT "localhost:9200/urunler_v2" -H 'Content-Type: application/json' -d'
{
  "settings": {
    "number_of_shards": 5,
    "number_of_replicas": 0,
    "refresh_interval": "-1"
  },
  "mappings": {
    "properties": {
      "urun_adi": {"type": "text", "analyzer": "turkish"},
      "fiyat": {"type": "scaled_float", "scaling_factor": 100}
    }
  }
}'

# 3. Asenkron reindex başlat
curl -X POST "localhost:9200/_reindex?wait_for_completion=false&requests_per_second=1000" 
  -H 'Content-Type: application/json' -d'
{
  "source": {"index": "urunler_v1"},
  "dest": {
    "index": "urunler_v2",
    "version_type": "external"
  },
  "conflicts": "proceed"
}'

# 4. Reindex tamamlandıktan sonra index ayarlarını geri al
curl -X PUT "localhost:9200/urunler_v2/_settings" -H 'Content-Type: application/json' -d'
{
  "refresh_interval": "1s",
  "number_of_replicas": 1
}'

# 5. Force merge ile segment sayısını optimize et
curl -X POST "localhost:9200/urunler_v2/_forcemerge?max_num_segments=1"

# 6. Atomic alias swap - bu işlem anlık gerçekleşir, kesinti olmaz
curl -X POST "localhost:9200/_aliases" -H 'Content-Type: application/json' -d'
{
  "actions": [
    {"remove": {"index": "urunler_v1", "alias": "urunler"}},
    {"add": {"index": "urunler_v2", "alias": "urunler"}}
  ]
}'

# 7. Her şey yolundaysa eski index'i sil
curl -X DELETE "localhost:9200/urunler_v1"

Bu akışta version_type: external kullanımına dikkat et. Bu, reindex sırasında source’daki _version değerini korur ve çakışma durumunda daha yüksek versiyonlu belgeyi kazanan yapar. Canlı sistemlerde bu davranış çoğunlukla istediğin şeydir.

Reindex Sonrası Doğrulama

İşlem bittikten sonra her zaman doğrulama yapmalısın. Belge sayılarını karşılaştırmak minimum gereksinim:

# Kaynak index belge sayısı
curl -X GET "localhost:9200/urunler_v1/_count"

# Hedef index belge sayısı
curl -X GET "localhost:9200/urunler_v2/_count"

# Index detaylı istatistikleri karşılaştır
curl -X GET "localhost:9200/urunler_v1/_stats/docs,store"
curl -X GET "localhost:9200/urunler_v2/_stats/docs,store"

# Rastgele örneklerle veri doğrulaması
curl -X GET "localhost:9200/urunler_v2/_search" -H 'Content-Type: application/json' -d'
{
  "query": {"match_all": {}},
  "size": 10,
  "sort": [{"_id": "asc"}]
}'

Belge sayıları tutmuyorsa birkaç olası neden var: Reindex sırasında source index’e yazılan yeni belgeler, version conflict nedeniyle atlanan belgeler veya script hatalarından dolayı skip edilen belgeler. Task API’nin döndürdüğü status alanındaki noops ve version_conflicts değerlerini kontrol et.

Büyük Veri Setleri için Slice ile Paralel Reindex

Milyonlarca belge içeren index’lerde Reindex’i hızlandırmak için slice kullanarak işlemi paralel çalıştırabilirsin.

# Slice 1/5
curl -X POST "localhost:9200/_reindex?wait_for_completion=false" 
  -H 'Content-Type: application/json' -d'
{
  "source": {
    "index": "buyuk_index",
    "slice": {"id": 0, "max": 5}
  },
  "dest": {"index": "buyuk_index_v2"}
}'

# Slice 2/5
curl -X POST "localhost:9200/_reindex?wait_for_completion=false" 
  -H 'Content-Type: application/json' -d'
{
  "source": {
    "index": "buyuk_index",
    "slice": {"id": 1, "max": 5}
  },
  "dest": {"index": "buyuk_index_v2"}
}'

# Kalan slice'lar için aynı pattern'ı tekrarla (id: 2, 3, 4)

Slice sayısını primary shard sayısına eşit veya onun katları olarak ayarlamak genellikle en iyi performansı verir. auto özelliğini de kullanabilirsin, bu durumda Elasticsearch slice sayısını otomatik belirler:

curl -X POST "localhost:9200/_reindex?wait_for_completion=false&slices=auto" 
  -H 'Content-Type: application/json' -d'
{
  "source": {"index": "buyuk_index"},
  "dest": {"index": "buyuk_index_v2"}
}'

Sık Karşılaşılan Hatalar ve Çözümleri

Socket timeout hatası: Büyük index’lerde işlem uzun sürer ve bağlantı kesilir. wait_for_completion=false kullanarak asenkron moda geç ve Task API ile takip et.

Heap pressure ve circuit breaker hataları: Batch size’ı düşür, throttle uygula ve destination index’te replica’yı sıfıra çek.

Version conflict: Canlı index’te reindex yapıyorsan conflicts: proceed ekle.

Remote reindex bağlantı hatası: reindex.remote.whitelist ayarının doğru yapıldığından ve elasticsearch servisinin yeniden başlatıldığından emin ol.

Disk alanı yetersizliği: Reindex sırasında hem source hem destination index disk kaplar. İşlem öncesinde en az source index boyutu kadar boş alan olduğunu kontrol et:

# Disk kullanımını kontrol et
curl -X GET "localhost:9200/_cat/allocation?v"
curl -X GET "localhost:9200/_cat/indices/urunler_v1?v&h=index,store.size,pri.store.size"

İzleme ve Loglama

Uzun süren Reindex işlemlerini izlemek için periyodik olarak task durumunu kontrol eden basit bir script yazabilirsin:

#!/bin/bash
TASK_ID="r1A2WoRbTwKZ516z6NEs5A:36619"
ES_HOST="localhost:9200"

while true; do
    DURUM=$(curl -s -X GET "${ES_HOST}/_tasks/${TASK_ID}" | 
        python3 -c "
import json,sys
d=json.load(sys.stdin)
t=d['task']['status']
print(f'Tamamlanan: {t["created"]}/{t["total"]} | Çakışma: {t["version_conflicts"]} | Throttle: {t["throttled_millis"]}ms')
")
    echo "$(date '+%H:%M:%S') - ${DURUM}"
    
    TAMAMLANDI=$(curl -s "${ES_HOST}/_tasks/${TASK_ID}" | 
        python3 -c "import json,sys; d=json.load(sys.stdin); print(d.get('completed', False))")
    
    if [ "$TAMAMLANDI" = "True" ]; then
        echo "Reindex tamamlandi!"
        break
    fi
    
    sleep 30
done

Sonuç

Elasticsearch Reindex API, veri göçü ve index yönetimi için güçlü ve esnek bir araç. Temel kullanım senaryolarından remote cluster göçüne, script ile dönüşümden alias tabanlı sıfır kesinti geçişine kadar pek çok durumu karşılıyor.

Üretim ortamında Reindex yaparken şu prensipleri unutma:

  • Her zaman alias kullan, uygulamanı direkt index adına bağlama
  • Reindex öncesi destination index’i hazırla ve replica’yı sıfıra çek
  • Büyük işlemlerde mutlaka wait_for_completion=false ve Task API kullan
  • Throttling ile cluster sağlığını izle, yoğun saatlerde hızı düşür
  • İşlem sonrası belge sayısı ve veri bütünlüğünü doğrula
  • Her zaman geri dönüş planın olsun, eski index’i hemen silme

Reindex işlemi ilk birkaç seferinde karmaşık gelebilir ama alias swap pattern’ını bir kez uyguladıktan sonra neden bu kadar zarif bir çözüm olduğunu anlarsın. Canlı sistemde sıfır kesinti ile index’i yeniden yapılandırmak, Elasticsearch’ü üretime alan her sysadmin’in bilmesi gereken kritik bir beceri.

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir