Elasticsearch "Unassigned Shard" Çözümü

Elasticsearch, son dönemde çok fazla popüler olmayan başlayan "Lucene" tabanlı full text search özelliği sağlayan bir veri tabanı sistemi. Büyük boyutlarda, çok sayıda json formatındaki verilerinizi buraya atıp, içeriklerinizi kolaylıkla indexleyip çok çeşitli uygulamalar geliştirebilirsiniz.

Son dönemde yaptığı atılımlara Elasticsearch adı altında bir çok özelliği barındıran başarılı bir veri tabanı sistemine evrilmiş durumda. Sağladığı başarılı REST API ile yönetimi ve monitor edilmesi oldukça kolay. Uygulamanıza text tabanlı arama özelliği katmak, büyük doküman verileri indexlemek istiyorsanız Apache Solr ile birlikte harika bir çözüm, hem de bedava indirerek hemence kullanabilirsiniz.

Kısa bir özetten sonra karşılaştığım bir soruna bulduğumuz basit ama etkili bir çözümü anlatalım. Elastic'de verileri "index" dediğimiz bloklar halinde tutar. Data kaybını ve sorgulamalarda hızını artırmak adına bu verimizi "replica" dediğimiz farklı sunucularda tutar. Biz de verimizi varsayılan olarak 2 kopya olarak tutuyoruz. Bazı indexlerimiz "unassigned shard" uyarı mesajı ile cluster'ımızı uyarı durumuna düşürdü. Bunun çözümü için farklı çözümler internette var. Benim işimi çözemese de onları inceleyebilirsiniz.

Benim bulduğum çözüm oldukça basit ve gayet güzel çalıştı :)

Öncelikle aşağıdaki kod ile sorunlu indexin replika sayısını yeniden set ederek yeni bir shard allocation başlattım. Bu işlem yeni uygun bir node üzerinde bu index'in yeni bir kopyasının oluşturulması demek.

PUT [index adı]/_settings
{
  "settings" : {"number_of_replicas": 2}
}

Daha sonra kopyalama işlemi bittiğinde "unassigned" olarak kalan shard'ın cluster'da uyarı olarak kalmasını önlemek için replika sayısnı 1 azaltarak her şeyi YEŞİL yaptım :) Bu işlem ile sorunlu görülen "unassigned" shard Elastic tarafından silindi. Benim sorunumu çözdü belki sizin de işinize yarar.

PUT [index adı]/_settings
{
  "settings" : {"number_of_replicas": 1}
}

Apache Spark - Cassandra

Bir önceki yazımızda kısaca Apache Spark'tan bahsetmiştik. Bu yazıda da gerçek dünyada kullanımına dair küçük bir kod parçası ve karşılacağımız olası bir problemin çözümünden bahsedeceğim.

Apache Cassandra'da yer alan büyük veri yığını tuttuğumuz tablomuza bağlanıp. Son 1 gün içinde tabloya insert edilmiş yaklaşık 4M kaydın "group by" ile "count (distinct)" sonucunu tek dosyasına yazan bir java uygulamasını aşağıdaki gibi yazabiliriz.

                 String sparkMaster = "local[4]";
   
    String cassandraNode = "x.x.x.x"; // veriyi çekeceğimiz cassandra node
 
    SparkConf conf = new SparkConf();
    conf.setAppName("Test Program");
    conf.setMaster(sparkMaster);
    conf.set("spark.cassandra.connection.host", cassandraNode);
    conf.set("spark.cores.max", "4");
    conf.set("spark.shuffle.service.enabled", "false");
    conf.set("spark.dynamicAllocation.enabled", "false");
    conf.set("spark.io.compression.codec", "snappy");
    conf.set("spark.rdd.compress", "true");
    conf.set("spark.driver.memory", "8G");
    conf.set("spark.executor.memory", "8G");


Spark işlemlerimizi koşacağımız uygulama ile önemli konfigürasyon ayarlarını yaptığımız kısım yukarıda. Kendi Cassandra cluster bilgilerinize ve uygulamanın koşacağı sunucunun kapasitesine göre ayarları değiştirebilirsiniz. Cassandra'dan büyük hacimde veri çekileceğinden en mantıklı yaklaşım Cassandra verisinin tutulduğu sunucu üzerinde bu kodu çalıştırmak olacaktır.

                JavaSparkContext sc = new JavaSparkContext(conf);
      
    SparkContextJavaFunctions functions =                     CassandraJavaUtil.javaFunctions(sc);
    JavaRDD<CassandraRow> rdd = functions.cassandraTable("[cassandra keyspace adı]", "[cassandra tablo adı]") .select("id").where("createDate > ?", sDate).where("createDate < ?", eDate).where("id > ?", 100);

Bu kısımda JavaSparkContext sınıfından bir örnek oluşturarak, bununla Cassandra'daki tablomuzdan belirldiğimiz tarih aralığında insert edilmiş, değeri 100'den büyük id değerine sahip kayıtları çekip rdd nesnemizin içine alıyoruz.

                rdd.saveAsTextFile("Test.txt");

Sonuçları text dosyasına kaydetmek tek satır ile mümkün. Tüm verimiz belirlediğimi dizinde dosyaya yazılıyor olacak.

Kodu Windows'ta çalıştırdığınızda aşağıdaki gibi bir hata alınması olası.

Could not locate executable null\bin\winutils.exe in the Hadoop binaries

Bu durumda aşağıdaki adımları takip ederek bu sorunu kolayca bertaraf edebiliriz.

1. http://public-repo-1.hortonworks.com/hdp-win-alpha/winutils.exe. adresinden uygulamayı indirin
2. Uygulamanın çalıştığı yerde C:\winutils\bin dizin yapısını oluşturun
3. Oluşturduğunuz dizine indirdiğiniz winutils.exe dosyasını kopyalayın.
4. Java kodunuza aşağıdaki satırı ekleyin.
5. Başka bir sorun yoksa başarılı bir şekilde işlem tamamlanmış demektir.

                  System.setProperty("hadoop.home.dir", "C:\\winutils");

       

Apache Spark İle Büyük Veri Analizi

Apache Spark en basit tanımıyla hızlı ve performanslı olarak "büyük veriyi" analiz etmek için kullanılan bir teknoloji. Daha çok Hadoop MapReduce ile karşılaştırılsa da aslında bazı noktalarda örtüşüp bazı yerlerde farklılaşırlar. Genel olarak aşağıdaki özelliklere sahip :

• MapReduce'den memory işlemlerinde 100x, disk işlemlerinde 10x daha hızlı
• Java, Scala, Python dil desteği
• Kolay kullanım
• Farklı veri kaynakları ile kolay entegrasyon
• Akış programlama (streaming), SQL ve komplex analitik desteği.

Yukarıda sayılan özelliklerden ötürü son zamanlarda popüler olmaya başlayan bir büyük veri analiz teknolojisi olarak öne çıkan Apache Spark'ı Apache Cassandra'ya attığımız büyük veri yığınları için de kullanmak mümkün. Datastax'ın ücretsiz olarak sunduğu connector kütüphanesi ile büyük veri ile dilediğiniz gibi sorgular koşup, çıktıları yeni bir tabloya yazabilir, her hangi bir dosyaya kaydedebilirsiniz. 

Örnek olarak günlük 5-6M kayıt attığınız, bir kaç yüz milyon satır veri içeren, Cassandra'da yer alan bir tablo düşünelim. NoSQL yapısı gereği her sorguyu SQL'deki gibi çalıştıramadığımızdan Apache Spark burada devreye giriyor. Cassandra'da yer alan tablomuza bağlanıp istediğimiz filtreye göre veri çekip bu veri üzerinde "count", "group by", "join" gibi NoSQL dünyasında istediğimiz gibi koşamadığımız fonksiyonları rahatça koşabiliriz. Spark temel olarak verimizi ilgili kaynaktan parça parça çekerek, dağıtık sistem üzerinde (spark cluster) sonuçları üreterek bize dilediğimiz çıktıyı üretir. Hadoop MapReduce'teki disk üzerinde gerçekleşen map ve reduce işlemleri, Spark tarafından memory'de gerçekleştiğinden çok daha hızlı sonuçlar alınır.

NoSQL Veritabanı Arayışımız

Teknolojinin hızla gelişmesiyle birlikte yeni ihtiyaçlar, yeni sorunlar da gün yüzüne çıkmaya başladı. İnternete bağlanma hızlarımızın artmasıyla birlikte ürettiğimiz veri miktarı inanılmaz oranda artış gösterdi. Bu baş döndürücü değişimlerden veri tabanı sistemleri de fazlasıyla nasibini aldı.

Daha önce Oracle, MySQL, MS Sql Server gibi veritabanları tüm piyasayı domine ederken gelişen farklılaşan ihtiyaçlarla yeni ürünler boy gösterdi. Bu bağlamda NoSQL dediğimiz, açılımı "Not only SQL", Türkçeye, "sql'in yeterli gelmediği nokta" gibi çevrilebilecek yeni bir kavram hayatımıza girdi. Oracle, MySQL gibi ilişkisel(RDBMS) sistemlerinin yetersiz kalmaya başladığı noktalarda NoSQL veritabanları imdada yetişti. HBase, Cassandra, MongoDB, Couchbase ve diğerleri...

Yakın döneme kadar MS SQL 2014 veritabanımız ile mutlu mesut yaşarken artan veri miktarımız, lisanslama maliyetleri, sistem bakım maliyetleri gibi nedenlerden ötürü bazı arayışlara girmek durumunda kaldık. Söz buraya gelmişken çoğu teknolojik gelişme son dönemde İnternet devlerinin mevcut çözümlerin yetersiz kaldığı noktada kendi iç kaynaklarıyla ihtiyaçlarına uygun farklı çözümler geliştirmeleriyle ortaya çıktı ve çıkmaya devam ediyor. Google'da geliştirilen Hadoop HDFS, Linkedin'de geliştirilip yayılan Apache Kafka gibi. Kendi hikayemize dönersek, SQL Server'da tuttuğumuz bir kaç tablomuz aşırı derecede büyüdü ve yaptığımız projeksiyonlarda yakın dönemde 2-3 katına çıkmasını öngörüyorduk. Tablo büyüklüğü 500M kayıt ve yaklaşık 600GB'lık bir veritabanı manasına geliyordu. Bu durum bize:

• Çok uzun süren backup süreleri
• Bakım çok zaman alan tablo bakım işlemleri (index rebuild, organize vs.)
• Artan olası lisanslama maliyetleri

gibi sorunlar ortaya çıkarmaya başladı. 

Geldiğimiz noktada yaptığımız araştırma ve incelemeler ile aşağıdaki özelliklere sahip bir veri tabanı sistemi ile sorunlarımızı aşabileceğimizi fark ettik.

• Açık lisanslı
• Dağıtık sistem olarak birden fazla sunucu üzerinde çalışabilecek
• Her hangi bir sunucu down olduğunda hizmet vermeye devam edecek (No Single Point of Failure)
• Çok pahalı olmayan ve eşit donanım özelliklerine sahip olmayan sunuculara üzerinde çalışabilecek
• Çok hızlı sorgulama ve insert etme özelliklerine sahip
• Yatay olarak çok rahat genişleyebilir
• Farklı programlama dilleri noktasında yardımcı kütüphanelere sahip olması

Bu özelliklerin yanında RDBMS sistemlerden alıştığımız bazı özelliklerden de feragat etmemiz gerekiyordu. NoSQL söz konusu olduğunda göz önüne almamız gereken bir noktada da RDBMS sistemlerin sağladığı aşağıdaki noktalardan kısmen yada tamamen vaz geçmeniz gerekliliği. Yoksa her şeyi halleden bir sistem olsa farklı ihtiyaçlara girmezdik değil mi :)

• ACID dediğimiz prensipleri desteklememeleri
• Tablolar arası ilişkileri RDBMS sistemlerdeki gibi tanımlayamamız. (join yapamamız)
• Transaction yapısının olmaması.

Yukarıdaki özellikleri ve eksiklikleri göz önüne aldığımızda ihtiyaçlarımıza uygun olabilecek en uygun NoSQL çözümünün Apache Cassandra olduğu ortaya çıktı. Tüm dünyada daha fazla kabul görmeye başlamış, binlerce büyük internet şirketinin kullandığı bu sistem sağladığı özellikler ile övgüyü hak ediyor. Başka yazılarda genel özelliklerini ve geçiş sürecimizi biraz daha detaylı anlatmaya çalışacağız.

Appache Cassandra Veri Aktarma

Her hangi bir RDBMS veri tabanı sisteminden Cassandra'ya geçişte karşılaşılan büyük çaplı sorunlardan biri de mevcut veriyi aktarmak. Cassandra söz konusu olduğunda en azından 100M/200M satıra sahip tablolardaki verinin aktarılması gerekir. Bunun için de geliştirilmiş bazı özel yaklaşımlar mevcut. Ben bu yazıda kendi yaşadığımın deneyimi aktarmak istiyorum.

Cassandra'ya bulk veri yüklemek için çeşitli yöntemler geliştirilmiş durumda. sstableloader Cassandra Bulk Loader ile ilgili bazı dokümanlar Datastax'ın sitesinde mevcut. Ancak bir miktar yazılım geliştirme yapılması gerekiyor. İnternette çok fazla kaynak da bulamadığınızdan sancılı bir süreç.

Bu genel probleme Brian Hess isimli bir geliştirici çok güzel ve kullanışlı bir ara yüz yazmış. Bununla ilgili Java programına bu linkten erişebilirsiniz. Kullanımı oldukça kolay ve çok stabil olduğunu düşünüyorum. Örneğin ben bir projemizde MS SQL 2014'te bulunan yaklaşık 500M kayda sahip bir tabloyu başarılı bir şekilde export ettim. Madde madde yapılanları özetlemek gerekirse :

• SQL Server'den tabloyu yaklaşık 60M-70M'luk kayıtlar içerecek şekilde csv formatında export ettim. Bununla ilgili bu yazıya göz atabilirsiniz.

• Cassandra-loader uygulamasını Java 8 yüklü bir sunucuya indirdim. Bunun için daha önce konfigüre konfigüre ettiğim Cassandra Cluster sunucularından birini kullandım. Ubuntu 14.04 sunucu üzerinde gayet stail bir şekilde çalıştığını söyleyebilirim. Aşağıdaki kod ile oluşturulan bu csv dosyaları sizin için Java Cassandra kütüphanesinin bulk insert fonksiyonları ile oldukça hızlı bir şekilde insert ediliyor.Buraya çok farklı parametreleri argüman olarak sağlayabilirsiniz. Bununla ilgili sitede açıklamalar mevcut. Örneğin aşağıda 45.32.xx.xx ip'de yer alan Cassandra sunucusu üzerinden MyKeyspace Cassandra veri tabanında yer alan MyTablo tablosuna her satırda 4 kolon olacak şekilde ve ayraç olarak TAB(\t) karakteri ile ayrılmış 60M satırlık csv dosyasını 10 paralel thread ile çalışacak şekilde insert etmesini, hata durumu oluşursa (eksik kolonlu satır vs.) 20000 hata oluşana dek bu işleme devam etmesini söylüyoruz. Dahasını ihtiyacınıza göre konfigüre edebilrisiniz.

./cassandra-loader -f /opt/cassandra-loader/MyTablo140to200M.csv -host 45.32.xx.xx -schema "MyKeyspace.MyTablo(col1, col2, col3, col4)" -delim "\t" -maxInsertErrors 20000 -maxErrors 20000 -numThreads 10

• Uygulama dosyadaki satırları 100K lık kümeler halinde insert ederek bunları ilerleme durumunu ekrana basıyor.

  

  İşlem Durumu

• İşlem logları da aynı dizinde .BADPARSE, .BADINSERT ve .LOG uzantılı dosyalar olarak tutuluyor.
• Benim çalışmalarımda gördüğüm kadarıyla 60M kayıt yaklaşık 30 dakikada tablolara insert edilmişti.Dosyaların her biri yaklaşık 10GB ve bunları 30dakikada insert edebilmek çok güzel :)

Cassandra'ya geçişlerin en zorlu kısımlarından biri büyük verilerin yeni ortama aktarılması işlemi. Henüz bu işlemi bir kaç tık ile yapabilmek mümkün olmadığından bir miktar çalışma yapmak gerekiyor. Bu işlemi kolaylaştırmak için yazılmış güzel bir uygulamanın işimizi ne kadar kolaylaştırdığını anlatmaya çalıştım. Sizin de tecrübelerinizi paylaşmanızı bekliyorum. Yeni yazıda görüşmek üzere.