Giriş
Öyle oldu ki şu anki iş yerimde bu teknolojiyle tanışmam gerekiyordu. Biraz arka planla başlayacağım. Bir sonraki toplantıda ekibimize entegrasyon oluşturmamız gerektiği söylendi. bilinen sistem. Entegrasyonla, bu iyi bilinen sistemin bize istekleri HTTP yoluyla belirli bir uç noktaya göndermesi ve tuhaf bir şekilde bizim de yanıtları bir SOAP mesajı biçiminde geri göndermemiz kastedildi. Her şey basit ve önemsiz görünüyor. Bundan, ihtiyacınız olduğu sonucu çıkıyor...
Görev
3 hizmet oluşturun. Bunlardan ilki Veritabanı Güncelleme Hizmetidir. Bu hizmet, üçüncü parti bir sistemden yeni veri geldiğinde veritabanındaki veriyi günceller ve CSV formatında bir dosya oluşturarak bir sonraki sisteme aktarır. İkinci hizmetin uç noktası, aktarılan dosyayı alan, doğrulayan ve FTP yoluyla dosya deposuna koyan FTP Aktarım Hizmeti olarak adlandırılır. Üçüncü hizmet olan tüketici veri aktarım hizmeti ise ilk ikisiyle eşzamansız olarak çalışıyor. Yukarıda tartışılan dosyayı almak için üçüncü taraf bir harici sistemden bir istek alır, hazır yanıt dosyasını alır, onu değiştirir (id, açıklama, linkToFile alanlarını günceller) ve yanıtı bir SOAP mesajı biçiminde gönderir. Yani genel tablo şu şekildedir: İlk iki hizmet ancak güncelleme verileri geldiğinde çalışmaya başlar. Üçüncü hizmet sürekli çalışır çünkü çok sayıda bilgi tüketicisi vardır, dakikada yaklaşık 1000 veri talebi vardır. Hizmetler sürekli olarak mevcuttur ve örnekleri test, demo, üretim öncesi ve üretim gibi farklı ortamlarda bulunur. Aşağıda bu hizmetlerin nasıl çalıştığını gösteren bir şema bulunmaktadır. Gereksiz karmaşıklığı önlemek için bazı detayların basitleştirildiğini hemen belirteyim.
Teknik Derinleşme
Soruna bir çözüm planlarken öncelikle Spring çerçevesini, Nginx dengeleyiciyi, Postgres veritabanını ve diğer teknik ve çok teknik olmayan şeyleri kullanarak Java'da uygulamalar yapmaya karar verdik. Teknik bir çözüm geliştirme zamanı, bu sorunu çözmeye yönelik diğer yaklaşımları değerlendirmemize olanak tanıdığından, bakışlarımız belirli çevrelerde moda olan Apache NIFI teknolojisine takıldı. Hemen söyleyeyim, bu teknoloji bu 3 hizmeti fark etmemizi sağladı. Bu yazıda tüketiciye dosya taşıma hizmeti ve veri aktarım hizmetinin geliştirilmesi anlatılacak ancak yazı faydalı olursa veritabanındaki verilerin güncellenmesi hizmeti hakkında da yazacağım.
Bu ne
NIFI, verilerin hızlı paralel yüklenmesi ve işlenmesi, kaynaklar ve dönüşümler için çok sayıda eklenti, konfigürasyonların sürümlendirilmesi ve çok daha fazlası için dağıtılmış bir mimaridir. Güzel bir bonus, kullanımının çok kolay olmasıdır. GetFile, sendHttpRequest ve diğerleri gibi önemsiz işlemler kareler olarak temsil edilebilir. Her kare, etkileşimi aşağıdaki şekilde görülebilecek bir süreci temsil eder. Süreç kurulumu etkileşimleri hakkında daha ayrıntılı belgeler yazılmıştır , Rusça konuşanlar için - . Belgeler, NIFI'nin nasıl paketten çıkarılıp çalıştırılacağının yanı sıra kareler olarak da bilinen süreçlerin nasıl oluşturulacağını mükemmel bir şekilde açıklamaktadır.
Bir makale yazma fikri, uzun bir araştırma ve alınan bilgilerin bilinçli bir şekilde yapılandırılmasının yanı sıra gelecekteki geliştiriciler için hayatı biraz daha kolaylaştırma arzusunun ardından doğdu.
Örnek
Karelerin birbirleriyle nasıl etkileşime girdiğine dair bir örnek ele alınmıştır. Genel şema oldukça basittir: Bir HTTP isteği alırız (Teorik olarak, isteğin gövdesinde bir dosya bulunur. NIFI'nin yeteneklerini göstermek için, bu örnekte istek, yerel dosya deposundan bir dosya alma sürecini başlatır.) ), ardından FH'den bir dosya alma işlemine ve ardından onu FTP yoluyla FH'ye taşıma işlemine paralel olarak isteğin alındığına dair bir yanıt göndeririz. Süreçlerin birbirleriyle akış dosyası adı verilen dosya aracılığıyla etkileşime girdiğini açıklığa kavuşturmak önemlidir. Bu, NIFI'da nitelikleri ve içeriği saklayan temel varlıktır. İçerik, akış dosyası tarafından temsil edilen verilerdir. Yani kabaca söylemek gerekirse, bir kareden bir dosya alıp diğerine aktarırsanız içerik sizin dosyanız olacaktır.
Gördüğünüz gibi bu resim genel süreci gösteriyor. HandleHttpRequest - istekleri kabul eder, ChangeText - bir yanıt gövdesi oluşturur, HandleHttpResponse - bir yanıt gönderir. FetchFile - dosya deposundan bir dosya alır, onu PutSftp karesine aktarır - bu dosyayı belirtilen adreste FTP'ye koyar. Şimdi bu süreç hakkında daha fazla bilgi.
Bu durumda istek her şeyin başlangıcıdır. Yapılandırma parametrelerine bakalım.
StandardHttpContextMap dışında buradaki her şey oldukça önemsizdir - bu, istek gönderip almanızı sağlayan bir tür hizmettir. Daha ayrıntılı olarak ve hatta örneklerle görebilirsiniz -
Şimdi karenin ChangeText yapılandırma parametrelerine bakalım. DeğiştirmeValue'ya dikkat etmeye değer - bu, kullanıcıya yanıt şeklinde iade edilecek olan şeydir. Ayarlarda günlük kaydı düzeyini ayarlayabilirsiniz, günlükleri görebilirsiniz {nerede nifi paketini açtığınız}/nifi-1.9.2/logs, ayrıca başarısızlık/başarı parametreleri de vardır - bu parametrelere dayanarak süreci bir bütün olarak düzenleyebilirsiniz . Yani, başarılı metin işleme durumunda, kullanıcıya yanıt gönderme işlemi çağrılacak ve başka bir durumda, başarısız olan işlemi basitçe günlüğe kaydedeceğiz.
Bir yanıtın başarıyla oluşturulduğu durum dışında, HandleHttpResponse özelliklerinde özellikle ilgi çekici hiçbir şey yoktur.
İsteği ve yanıtı sıraladık; dosyayı alıp FTP sunucusuna yerleştirmeye geçelim. FetchFile - ayarlarda belirtilen yolda bir dosya alır ve onu bir sonraki işleme aktarır.
Ve sonra PutSftp karesi dosyayı dosya deposuna yerleştirir. Aşağıda konfigürasyon parametrelerini görebiliriz.
Her karenin başlatılması gereken ayrı bir süreç olmasına dikkat etmekte fayda var. Herhangi bir karmaşık özelleştirme gerektirmeyen en basit örneğe baktık. Daha sonra, oluklara biraz yazacağımız sürece biraz daha karmaşık bakacağız.
Daha karmaşık örnek
SOAP mesajını değiştirme süreci nedeniyle tüketiciye veri aktarım hizmetinin biraz daha karmaşık olduğu ortaya çıktı. Genel süreç aşağıdaki şekilde gösterilmektedir.
Burada fikir de pek karmaşık değil: Tüketiciden verilere ihtiyacı olduğuna dair bir talep aldık, bir mesaj aldığına dair bir yanıt gönderdik, yanıt dosyasını alma sürecini başlattık, ardından onu belirli bir mantıkla düzenledik ve ardından dosyayı sunucuya SOAP mesajı şeklinde tüketiciye aktardı.
Yukarıda gördüğümüz kareleri tekrar anlatmaya gerek yok sanırım, hemen yenilerine geçelim. Herhangi bir dosyayı düzenlemeniz gerekiyorsa ve sıradan ChangeText türü kareler uygun değilse, kendi komut dosyanızı yazmanız gerekecektir. Bu, ExecuteGroogyScript karesi kullanılarak yapılabilir. Ayarları aşağıda sunulmuştur.
Komut dosyasını bu kareye yüklemek için iki seçenek vardır. Birincisi, komut dosyası içeren bir dosyayı indirmektir. İkincisi, scriptBody'ye bir komut dosyası eklemektir. Bildiğim kadarıyla, executiveScript karesi birkaç dili destekliyor - bunlardan biri harika. Java geliştiricilerini hayal kırıklığına uğratacağım - bu tür karelere Java'da komut dosyaları yazamazsınız. Gerçekten isteyenler için kendi özel karenizi oluşturup NIFI sistemine eklemeniz gerekiyor. Tüm bu operasyona, bu yazıda ele almayacağımız tef ile oldukça uzun bir dans eşlik ediyor. Harika bir dil seçtim. Aşağıda, bir SOAP mesajındaki kimliği basitçe artımlı olarak güncelleyen bir test komut dosyası bulunmaktadır. Bunu not etmek önemlidir. Dosyayı flowFile'dan alıp güncelliyorsunuz, güncel olarak oraya geri koymanız gerektiğini unutmayın. Ayrıca tüm kütüphanelerin dahil edilmediğini de belirtmek gerekir. Hala kütüphanelerden birini içe aktarmanız gerekebilir. Diğer bir dezavantaj ise bu karedeki betiğin hata ayıklamasının oldukça zor olmasıdır. NIFI JVM'ye bağlanmanın ve hata ayıklama sürecini başlatmanın bir yolu var. Şahsen ben yerel bir uygulama başlattım ve oturumdan bir dosya almayı simüle ettim. Ayrıca yerel olarak hata ayıklama da yaptım. Bir komut dosyası yüklenirken ortaya çıkan hatalar Google için oldukça kolaydır ve NIFI'nın kendisi tarafından günlüğe yazılır.
import org.apache.commons.io.IOUtils
import groovy.xml.XmlUtil
import java.nio.charset.*
import groovy.xml.StreamingMarkupBuilder
def flowFile = session.get()
if (!flowFile) return
try {
flowFile = session.write(flowFile, { inputStream, outputStream ->
String result = IOUtils.toString(inputStream, "UTF-8");
def recordIn = new XmlSlurper().parseText(result)
def element = recordIn.depthFirst().find {
it.name() == 'id'
}
def newId = Integer.parseInt(element.toString()) + 1
def recordOut = new XmlSlurper().parseText(result)
recordOut.Body.ClientMessage.RequestMessage.RequestContent.content.MessagePrimaryContent.ResponseBody.id = newId
def res = new StreamingMarkupBuilder().bind { mkp.yield recordOut }.toString()
outputStream.write(res.getBytes(StandardCharsets.UTF_8))
} as StreamCallback)
session.transfer(flowFile, REL_SUCCESS)
}
catch(Exception e) {
log.error("Error during processing of validate.groovy", e)
session.transfer(flowFile, REL_FAILURE)
}Aslında meydanın kişiselleştirilmesi burada bitiyor. Daha sonra güncellenen dosya, dosyayı sunucuya göndermekten sorumlu olan square'e aktarılır. Aşağıda bu square'in ayarları verilmiştir.
Bir SOAP mesajının iletileceği yöntemi açıklıyoruz. Nerede olduğunu yazıyoruz. Daha sonra bunun SABUN olduğunu belirtmeniz gerekir.
Ana bilgisayar ve eylem (soapAction) gibi çeşitli özellikler ekleyin. Kaydedip kontrol ediyoruz. SOAP isteklerinin nasıl gönderileceği hakkında daha fazla ayrıntı görebilirsiniz
NIFI süreçlerini kullanmak için çeşitli seçeneklere baktık. Nasıl etkileşime giriyorlar ve gerçek faydaları nedir? Ele alınan örnekler test amaçlıdır ve savaşta gerçekte olanlardan biraz farklıdır. Bu makalenin geliştiriciler için biraz faydalı olacağını umuyorum. İlginiz için teşekkür ederiz. Sorularınız olursa yazın. Cevaplamaya çalışacağım.
Kaynak: habr.com