Python çoklu sürecini değiştirin! Berkeley açık kaynak dağıtılmış çerçeve Ray
Paralel ve dağıtılmış bilgi işlem, modern uygulamaların ana içeriğidir. Uygulamaları hızlandırmak veya uygun ölçekte çalıştırmak için birden çok çekirdek veya birden çok makine kullanmamız gerekiyor. Web tarayıcıları ve arama tarafından kullanılan altyapı, birinin dizüstü bilgisayarında çalışan tek iş parçacıklı bir program değil, birbiriyle iletişim kuran ve etkileşimde bulunan bir dizi hizmettir.
Bulut bilişim, tüm boyutlarda (bellek, bilgi işlem, depolama vb.) Sınırsız ölçeklenebilirlik vaat ediyor. Bu vaadi gerçekleştirmek, bulut programlama ve dağıtılmış uygulamalar oluşturmak için yeni araçlar gerektirir.
Neden Ray kullanıyorsunuz?
Pek çok eğitim, Python'un çoklu işlem modülünün (https://docs.python.org/2/library/multiprocessing.html) nasıl kullanılacağını açıklar. Ne yazık ki, çok işlemli modülün modern uygulamaların gereksinimlerini ele alırken ciddi eksiklikleri vardır. Bu gereksinimler şunları içerir:
- Aynı kodu birden çok bilgisayarda çalıştırın.
- Durum bilgisi olan ve iletişim kurabilen mikro hizmetler ve aktörler oluşturun.
- Makine arızalarını zarafetle halledin.
- Büyük nesneleri ve sayısal verileri etkili bir şekilde işleyin.
Ray (https://github.com/ray-project/ray) tüm bu sorunları çözer ve basitliği korurken karmaşık davranışları mümkün kılar.
Gerekli kavramlar
Geleneksel programlama iki temel konsepte dayanır: işlevler ve sınıflar. Bu yapı taşları kullanılarak sayısız uygulama oluşturulabilir.
Ancak uygulamaları dağıtılmış bir ortama taşıdığımızda, bu kavramlar genellikle değişir.
Bir yandan, OpenMPI, Python çoklu işlem ve ZeroMQ gibi araçlar, mesaj göndermek ve almak için düşük seviyeli temeller sağlar. Bu araçlar çok güçlüdür, ancak farklı soyutlamalar sağlarlar, bu nedenle bunları kullanmak için tek iş parçacıklı uygulamaları sıfırdan yeniden yazmanız gerekir.
Öte yandan, model eğitimi için TensorFlow, veri işleme ve SQL desteği için Spark ve akış için Flink gibi etki alanına özgü bazı araçlarımız da var. Bu araçlar sinir ağları, veri kümeleri ve akışlar gibi daha üst düzey soyutlamalar sağlar. Bununla birlikte, seri programlama için kullanılan soyutlamalardan farklı oldukları için, bunları kullanmak için uygulamanın sıfırdan yeniden yazılması gerekir.
Dağıtılmış bilgi işlem için araçlar
Ray, benzersiz bir orta zemine sahiptir. Yeni kavramlar tanıtmaz, ancak işlev ve sınıf kavramlarını kullanır ve bunları dağıtılmış görevlere ve aktörlere dönüştürür. Ray, seri uygulamaları büyük değişiklikler yapmadan paralel hale getirebilir.
Ray'i kullanmaya başlayın
Ray.init () komutu, ilgili tüm Ray işlemlerini başlatacaktır. Kümeye geçerken, değiştirilmesi gereken tek satır budur (küme adresini geçmemiz gerekir). Bu süreçler şunları içerir:
- Python işlevlerini paralel olarak çalıştıran birçok çalışan işlem vardır (her CPU çekirdeği için yaklaşık bir işçi).
- İşçilere (ve diğer bilgisayarlara) "görevler" atamak için kullanılan bir zamanlayıcı işlemi. Görev, Ray tarafından planlanan ve bir işlev çağrısına veya yöntem çağrısına karşılık gelen bir iş birimidir.
- Çalışanlar arasında nesneleri verimli bir şekilde paylaşmak için paylaşılan bellek nesnesi depolama kitaplığı (kopya oluşturmaya gerek yoktur).
- Bellek içi veritabanı, bir makine arızası durumunda görevleri yeniden çalıştırmak için gereken meta verileri depolamak için kullanılır.
Ray işçisi bağımsız bir süreçtir, iş parçacığı değil.Python'da global bir yorumlayıcı kilidi olduğundan, çoklu okuma desteği çok sınırlıdır.
Paralel görev
Python f işlevini bir "uzak işleve" (uzaktan ve eşzamansız olarak çalıştırılabilen bir işlev) dönüştürmek için, bu işlevi bildirmek için @ ray.remote dekoratörünü kullanabilirsiniz. Daha sonra f.remote () işlev çağrısı hemen bir gelecek döndürecektir (gelecek, son çıktıya bir referanstır) ve gerçek işlev yürütmesi arka planda gerçekleştirilecektir (bu işlevi bir görev yürütmeye çağırıyoruz).
Kodu kopyala
ışın almak ithalat zamanı # Ray'i başlatın. ray.init () @ ray.remote def f (x): zaman.sleep (1) dönüş x # 4 görevi paralel olarak başlatın. result_ids = aralıktaki i için (4): result_ids.append (f.remote (i)) # Görevlerin tamamlanmasını ve sonuçları almasını bekleyin. # En az 4 çekirdek ile bu 1 saniye sürecektir. sonuçlar = ray.get (sonuç_ids) #Python'da paralel görevleri çalıştırmak için kodlama
F.remote (i) çağrısı hemen döndüğünden, bu kod satırının dört kez çalıştırılması, f'nin dört kopyasını paralel olarak çalıştırabilir.
Görev bağımlılığı
Bir görev, diğer görevlere de bağlı olabilir. Aşağıdaki kodda, multiply_matrices görevi iki create_matrix görevinin çıktısına dayanmaktadır, bu nedenle ilk iki görev yürütülmeden önce çalıştırılmayacaktır. İlk iki görevin çıktısı otomatik olarak üçüncü göreve bir parametre olarak geçirilecek ve gelecek ilgili değerle değiştirilecektir. Bu şekilde, görevler herhangi bir DAG bağımlılığına göre birlikte gruplanabilir.
Kodu kopyala
numpy'yi np olarak içe aktar @ ray.remote def create_matrix (boyut): return np.random.normal (size = size) @ ray.remote def multiply_matrices (x, y): dönüş np.dot (x, y) x_id = create_matrix.remote () y_id = create_matrix.remote () z_id = multiply_matrices.remote (x_id, y_id) # Sonuçları alın. z = ray.get (z_id)Üç görev vardır, üçüncü görev ilk iki görevin çıktısına bağlıdır.
Değerleri verimli bir şekilde toplayın
Görev bağımlılıklarını daha karmaşık şekillerde kullanabiliriz. Örneğin, 8 değeri bir araya toplamak istediğimizi varsayalım. Örneğimizde tamsayı toplama gerçekleştireceğiz, ancak birçok uygulamada büyük vektörleri birden çok bilgisayarda toplamak performans darboğazlarına neden olabilir. Şu anda, kümelemenin çalışma süresini doğrusaldan logaritmik düzeye, yani toplanan değerlerin sayısını azaltmak için bir kod satırını değiştirin.
Soldaki bağımlılık grafiğinin derinliği 7, sağdaki bağımlılık grafiğinin derinliği 3'tür. Hesaplama aynı sonucu verir, ancak sağdaki bağımlılık grafiği daha hızlı çalışır.
Yukarıda bahsedildiği gibi, bir görevin çıktısını sonraki göreve girdi olarak sağlamak için, yalnızca birinci görevin bir parametre olarak döndürdüğü geleceği ikinci göreve geçirin. Ray'in zamanlayıcısı, görev bağımlılıklarını otomatik olarak değerlendirir. İkinci görev, ilk görev tamamlanana kadar yürütülmeyecek ve ilk görevin çıktısı, ikinci görevi gerçekleştiren makineye otomatik olarak gönderilecektir.
Kodu kopyala
ithalat zamanı @ ray.remote def ekle (x, y): zaman.sleep (1) x + y döndür # Değerleri yavaşça toplayın. Bu yaklaşım O (n) alır, burada n, # Toplanan değer sayısı. Bu durumda, 7 saniye. id1 = add.remote (1, 2) id2 = add.remote (id1, 3) id3 = add.remote (id2, 4) id4 = add.remote (id3, 5) id5 = add.remote (id4, 6) id6 = add.remote (id5, 7) id7 = add.remote (id6, 8) sonuç = ray.get (id7) # Değerleri ağaç yapılı bir modelde toplayın. Bu yaklaşım # O (log (n)) alır, bu durumda 3 saniye. id1 = add.remote (1, 2) id2 = add.remote (3, 4) id3 = add.remote (5, 6) id4 = add.remote (7, 8) id5 = add.remote (id1, id2) id6 = add.remote (id3, id4) id7 = add.remote (id5, id6) sonuç = ray.get (id7)Doğrusal bir şekilde değerlerin toplanmasının karşılaştırılması ve bir ağaç yapısındaki değerlerin toplanması
Yukarıdaki kod çok açık, ancak lütfen her iki yöntemin de daha kısa olan bir süre döngüsü kullanılarak uygulanabileceğini unutmayın.
Kodu kopyala
# Yavaş yaklaşım. değerler = while len (değerler) > 1: değerler = + değerler sonuç = ray.get (değerler) # Hızlı yaklaşım. değerler = while len (değerler) > 1: değerler = değerler + sonuç = ray.get (değerler)Toplama için daha kısa bir uygulama şeması. İki kod bloğu arasındaki tek fark, "add.remote" çıktısının listeden önce mi yoksa sonra mı yerleştirildiğidir.
Sınıftan oyuncuya
Dağıtık bir ortamda bile sınıfları kullanmadan ilginç uygulamalar geliştirmek zordur.
@ Ray.remote dekoratörünü kullanarak bir Python sınıfı bildirebilirsiniz. Ray, bir sınıfı başlatırken, kümede çalışan ve sınıf nesnesinin bir kopyasını tutan bir süreç olan yeni bir "aktör" oluşturur. Bu aktöre yapılan yöntem çağrıları, aktör sürecinde çalışan görevlere dönüştürülür ve aktörün durumuna erişilebilir ve değiştirilebilir. Bu şekilde, değişken durum, uzak işlevlerle mümkün olmayan birden fazla görev arasında paylaşılabilir.
Her aktör, yöntemleri sırayla yürütür (her yöntem atomik bir yöntemdir), bu nedenle yarış koşulu yoktur. Paralellik, birden çok aktör yaratarak elde edilebilir.
Kodu kopyala
@ ray.remote sınıf Sayacı (nesne): def __init __ (öz): self.x = 0 def inc (kendi): self.x + = 1 def get_value (öz): self.x'e dön # Bir aktör süreci oluşturun. c = Counter.remote () # Oyuncunun sayaç değerini kontrol edin. print (ray.get (c.get_value.remote ())) # 0 # Sayacı iki kez artırın ve değeri tekrar kontrol edin. c.inc.remote () c.inc.remote () print (ray.get (c.get_value.remote ())) # 2Python sınıfını aktör olarak örnekleme
Yukarıdaki örnek, aktörün en basit kullanımıdır. Counter.remote (), Counter nesnesinin bir kopyasını tutan yeni bir aktör süreci oluşturur. C.get_value.remote () ve c.inc.remote () 'a yapılan çağrılar, uzak aktör sürecinde görevler gerçekleştirecek ve aktörün durumunu değiştirecektir.
aktör kolu
Yukarıdaki örnekte, aktörün yöntemini yalnızca ana Python betiğinde çağırıyoruz. Bir aktörün en güçlü yönlerinden biri, diğer aktörlerin veya diğer görevlerin aynı oyuncunun yöntemlerini çağırabilmesi için ona bir yön verebilmemizdir.
Aşağıdaki örnek, mesajları kaydedebilen bir aktör oluşturur. Birkaç işçi görevi, aktörlere tekrar tekrar mesaj gönderir ve ana Python betiği, mesajları periyodik olarak okur.
Kodu kopyala
ithalat zamanı @ ray.remote class MessageActor (nesne): def __init __ (öz): self.messages = def add_message (öz, mesaj): self.messages.append (mesaj) def get_and_clear_messages (self): mesajlar = self.messages self.messages = mesajları iade et # Etrafında dönen ve iten bir uzaktan işlev tanımlayın Oyuncuya # mesaj. @ ray.remote def işçi (mesaj_aktör, j): aralıktaki i için (100): zaman.sleep (1) message_actor.add_message.remote ( "{} Adlı aktörden {} mesajı.". Format (i, j)) # Bir mesaj aktörü oluşturun. message_actor = MessageActor.remote () # Oyuncuya mesaj gönderen 3 görev başlatın. # Mesajları periyodik olarak alın ve yazdırın. Aralık içindeki _ için (100): new_messages = ray.get (message_actor.get_and_clear_messages.remote ()) print ("Yeni mesajlar:", yeni_mesajlar) zaman.sleep (1) # Bu komut dosyası aşağıdakine benzer bir şey yazdırır: # Yeni mesaj: # Yeni mesaj: # Yeni mesaj: # Yeni mesaj: # Yeni mesaj: # Yeni mesaj: # Yeni mesaj:Birden çok eşzamanlı görevde aktörü çağırma yöntemi
Oyuncu çok güçlü. Python sınıflarını mikro hizmetler olarak başlatmak ve bu mikro hizmeti diğer aktörlerden ve görevlerden (hatta diğer uygulamalardan) sorgulamak için kullanabilirsiniz.
Görevler ve aktörler, Ray tarafından sağlanan temel soyutlamalardır. Bu iki kavram çok yaygındır ve Ray'in takviye öğrenme, hiperparametre ayarı ve Pandaların hızlandırılması için yerleşik kitaplıkları dahil olmak üzere karmaşık uygulamaları uygulamak için kullanılabilir.
Orijinal İngilizce:
https://medium.com/@robertnishihara/modern-parallel-and-distributed-python-a-quick-tutorial-on-ray-99f8d70369b8
