Amazon Kinesis Kullanımı

Merhaba arkadaşlar, bu dersimizde gerçek zamanlı akış verilerinin toplanmasında, işlenmesinde; hızlı, ölçeklenebilir olan Amazon Kinesis servisini inceleyeceğiz.
Akış verileri sürekli olarak farklı kaynaklardan oluşturulan verilerdir. Özellikle sosyal medyanın, endüstri 4.0 ile hayatımıza giren nesnelerin interneti ile birlikte büyük data kaynaklarının çoğalmasıyla oldukça önemli bir konu haline gelmiştir.

Bu verilere bir web sitesinin log dosyaları, e-ticaret sitesindeki sayfa analitikleri, sosyal medyadaki paylaşım verileri gibi bir çok örnek gösterebiliriz. Sistemlerin trafiği arttığında bekleneceği üzere bu verilerin büyüklüğü ve önemi de aynı oranda artmaktadır.

Bu sebeple bulut servisler üzerinde Streaming Data hizmeti sunulması servislerin ölçeklenmesinde ve gerçek zamanlı veri akışlarının sekteye uğramaması için önemli hale gelmektedir.

Amazon Kinesis içerisinde dört farklı alt servis bulundurmaktadır. Alt servisler; Kinesis Video Streams, Kinesis Data Streams, Kinesis Data FireHose ve Kinesis Data Analytics servislerinden oluşmaktadır.

Yazımızın devamında Kinesis servislerinin ne olduğunu ve hangi alanlarda kullanıldıklarını inceleyeceğiz.

Kinesis Video Streams

Görüntü cihazlarından güvenli bir şekilde AWS servisine videoların stream edilerek; AWS içerisinde bulunan çeşitli makine öğrenmesi ve yapay zeka servisleri ile işlenmesine veya kendi ihtiyacınıza uygun bir şekilde videoların işlenmesini sağlar.

Kinesis Video Stream’e her türlü cihazdan(akıllı telefonlar,güvenlik kameraları,dronelar,uydular vb…) güvenli bir şekilde videoları alabilmektedir. Bu verileri depolanması için Amazon S3 servisi kullanmaktadır ve aktarılan tüm verilerin şifreli bir şekilde sağlanmasına olanak sağlar.

Örnek olarak güvenlik kameraları görüntülerini Kinesis Video Streams aracılığı ile yine AWS yapay zeka servisleri yardımıyla istenilen plakanın tespiti yapılabilir. Çeşitli medya kaynaklarındaki canlı görüntülerin Kinesis Video Streams aracılığı ile AWS yapay zeka servisleri aracılığı ile canlı bir şekilde medya analizi, konuşmaların yazıya dökülmesi, durum analizi gibi çeşitli analizlerin yapılmasına olanak sağlar.

Detaylı bilgileri AWS’nin Türkçe hazırladığı bu linkten bulabilirsiniz.

Kinesis Data Streams

Data Streams artan iş yüklerine göre kolayca ölçeklenebilen bir veri akış sistemidir. Bu sistem sayesinde neredeyse her türlü kaynaktan gelen verilerin hiç bir sorun yaşanmadan depolanması ve hızlı bir şekilde analizlerinin yapılması sağlanmaktadır. Kinesis Data Streams ile saniyede gigabytelarca verilerin akışı sağlanıp neredeyse saniyelerden daha kısa sürede analizleri yapılabilir. Bu sayede anlık trafik artışları veya beklenmedik olaylarda hiç sorun yaşanmadan ölçeklenmesiyle servislerinizin devamlılığını sağlamasına yardımcı olur.

Data Streams içerisinde bulunan verilerin şifreli iletimleri ve VPC aracılığı ile ulaşarak güvenli bir şekilde verilerin aktarılması ve işlenmesi sağlanabilmektedir. Data Streams servisindeki verileri Lambda,EC2 gibi servisleri kullanarak işemler yapılabilmektedir.

Örnek olarak 2017 yılında Netflix tarafından AWS blogunda yayımlanan yazıyı kullanarak Netflix’in Kinesis Kullanımını gösterebiliriz. Netflix AWS üzerindeki 100.000’lerce instancelarında çalışan servislerin metriklerini ve çeşitli trafik bilgilerini Kinesis Data Streams kullanarak toplamakta ve analiz etmektedir.İlgili blog yazısına buradan erişebilirsiniz.

Aynı şekilde binlerce IOT cihazından gelen verilerin servislerinize hiç bir ekstra yük çıkarmadan işlenmesi ve analizlerinin yapılması da Data Streams kullanarak gerçekleştirilebilir.

Data Stream için daha detaylı bilgiyi AWS’nin Türkçe hazırladığı sayfadan edinebilirsiniz.

Kinesis Data Firehose

Firehose ile binlerce farklı kaynaktan gelen akış verilerinin toplanması ve gerekiyorsa dönüştürülüp S3, Elasticsearch, Redshift veya Splunk’a gönderimini kolay ve hızlı bir şekilde yapılmasını sağlayan sistemdir. Bu sayede sizin servislerinizin aktarım hızına erişecek seviyede ölçeklenebilen Firehose, gönderilen akış verilerinin S3’te depolanmasından önce Apache Parquet veya Apache ORC biçimlerine otomatik olarak dönüştürülebilmektedir. Bu sayede bu dönüşümü yapması için ek bir servis kullanımına ihtiyaç duyulmamaktadır.

Örnek olarak çalışmakta olan sunucularımızdaki çeşitli log dosyalarını Firehose ile Splunk veya Amazon Elasticsearch servislerine göndererek sistemlerimiz için gerekli güvenlik analizlerini, çeşitli alarm tanımlarını ve olay anında rahatça bilgiye erişmeyi kolaylaştırabiliriz. Bir e-ticaret sitesindeki ödeme servisine ait operasyonel logların Elasticsearch servisine gönderilip zararlı davranışların fark edilmesini de örnek olarak gösterebiliriz.

Firehose için daha detaylı bilgiyi AWS’nin Türkçe hazırladığı sayfaya ulaşarak okuyabilirsiniz.

Kinesis Data Analytics

Data Analytics ile veri akışlarını ek bir servis kullanmadan Java veya SQL ile canlı olarak işlenmesini sağlamaktadır. Firehose veya Data Streams akışlarından gelen verilerin sorgulanıp, bulunan sonuçlarda istenilen değişiklik veya analizlerin yapılmasını sağlar. Bu sayede farklı amaçlarla tutulması gereken verilerin, ilgili servislere aktarılmadan çeşitli alanlarının silinmesi, değiştirilmesi veya yeni bir alanın oluşturulması mümkün olur. Bu sayede bu işlemler için ek bir servis kurulmadan hızlıca geliştirilme yapılmasına olanak sağlar.

Apache Flink baz alan çeşitli Java kütüphanelerini kullanarak akış verilerinin analizi sağlanır. Aynı şekilde SQL ile istenen analizlerin gerçekleştirilmesi de oldukça kolaydır.

Örnek olarak canlı borsa verilerini alan bir Data Analytics üzerinde makine öğrenmesi algoritmalarıyla canlı olarak işlem analizlerinin yapılmasını örnek gösterebiliriz.

Örnek

Örnek olarak bir Data Stream oluşturalım ve bu stream içerisine veri gönderip, bu veriye ulaşalım.

Öncelikle web üzerinden bir Data Streams oluşturuyoruz.

Number of open shards : Saniyede yazılacak verinin miktarına göre Shard sayısını belirlememize yarar. Bu özelliği kullanarak veri akışlarımızın artışına veya azalışlarına göre ölçeklendirebiliriz.

Örnekte görüleceği üzere ben 1 Shard seçtiğim için saniyede 1 MiB yazma, 2MiB okuma hızına sahip oldu.

Aynı sayfada bulunan Shard Estimator yardımıyla ne kadar shard ihtiyacınız olacağını hesaplayabilirisiniz.

Kinesis Streamlerinde verilere eriştikten sonra silebilmemiz mümkün olmuyor ancak verinin Kinesiste erişebilir kalma süresini aşağıdaki gibi değiştirebiliriz. Bu özellik sayesinde streamdeki verileri analiz eden servisimizde bir sıkıntı meydana gelirse, belirli bir noktadan itibaren analizi bir daha çalıştırabilme imkanımız olabilir.

Kinesis verilerimizi base64 formatında beklediği için base64 dönüşümünü yapalım.

$echo test verimiz | base64 
dGVzdCB2ZXJpbWl6Cg==

Put Record komutunu kullanarak bir veri gönderelim. Bu komut stream adı, veri ve partition-key parameterlerini almaktadır. Partition Key parametresiyle shard içerisindeki verileri gruplamak için kullanılabilir.

$ aws kinesis put-record --stream-name aws-ders-stream --data dGVzdCB2ZXJpbWl6Cg== --partition-key ornekpartitionkey
{
    "ShardId": "shardId-000000000000",
    "SequenceNumber": "49607156334460057456100334565168052496218190752193183746"
}

Recordları getirmeden önce bir iteratöre ihtiyacımız bulunmaktadır. Iterator ile shard ve tip belirterek verilere ulaşabiliriz.

Iterator tipi olarak SequenceNumber, Timestamp veya Latest girerek iteratörümüzü değiştirebiliriz. Bu sayede Latest tipiyle hep en son verileri alabiliriz ya da Kinesis Streamlerini işleyen servisimizde bir sorun çıktıysa doğru şekilde çalışan son sequence number girerek kaldığı yerden devam etmesini sağlayabiliriz.

$ aws kinesis get-shard-iterator --stream-name aws-ders-stream --shard-id shardId-000000000000 --shard-iterator-type AT_SEQUENCE_NUMBER --starting-sequence-number 49607156334460057456100334565168052496218190752193183746
{
    "ShardIterator": "AAAAAAAAAAH3583jaXNIdCNx+urR5qs3HnANST2zQzqGPo/Sgv5yWhSxz19O7AQ9jfTr9doADZzmYu4dZZytOWI5L89lzptmtDwkoxbE/4iWiWvDEs0gHaZv1dXcgrE5IZZ5FADNQtJNtE5DishLrgcky2dHeVU7gKGtqwLXc2pn5rtAu7iXAwMlGo30P2g41hX0WzVg1mEv/ixyO5ISdcXpS+EZ0WNo"
}

Şimdi ise bu iterator ile verimize ulaşabiliriz. Aşağıdaki sonuçtanda görülebileceği gibi bir sonraki iterator alanındaki veriyle kayıtlar almaya devam edebiliriz.

$ aws kinesis get-records --shard-iterator AAAAAAAAAAH3583jaXNIdCNx+urR5qs3HnANST2zQzqGPo/Sgv5yWhSxz19O7AQ9jfTr9doADZzmYu4dZZytOWI5L89lzptmtDwkoxbE/4iWiWvDEs0gHaZv1dXcgrE5IZZ5FADNQtJNtE5DishLrgcky2dHeVU7gKGtqwLXc2pn5rtAu7iXAwMlGo30P2g41hX0WzVg1mEv/ixyO5ISdcXpS+EZ0WNo
{
    "Records": [
        {
            "SequenceNumber": "49607156334460057456100334565168052496218190752193183746",
            "ApproximateArrivalTimestamp": "2020-05-19T21:02:49.229000+03:00",
            "Data": "dGVzdCB2ZXJpbWl6Cg==",
            "PartitionKey": "ornekpartitionkey"
        }
    ],
    "NextShardIterator": "AAAAAAAAAAFJ0Zhz5TTVxSaxz9rPfdSsvTF7uHqF7d2k6aMjbTVO8c00/FqhA5J1h3pg5qc4mzOsCMum7NyJbbOQtkzCzg7Tr/3sFUI2Nt/zjwpFbOMNkGQBi86Z+7RzxNqDKg19a3lnH+3qBElrl6PyRjFnluscG1LNhshK/YRw3gBZYWX38BSuDgckCQQjBLoTgH4TWv+74YglZlxp3JeFq+Mid+Pt",
    "MillisBehindLatest": 0
}

Kinesis için işlemlerin daha kolay yapılması amacıyla Kinesis Client Library kullanabiliriz. KCL hakkında detaylı bilgiye buradan ulaşabilirsiniz.

Bu yazımızda Amazon Kinesis kullanarak video ve veri akışlarının toplanmasını, anlık olarak işlenmesini ve analiz edilebilmesini işledik. Kinesis yardımıyla veri akışları için ekstra bir servis yönetmeden ve akış verilerimizin hızına göre ölçeklenmesiyle verileri hızlıca analiz edip gerekli işlemlerimizi gerçekleştirebiliriz.