Spark Akışı Akışı İşleme Projesi 5-Spark Akışı ile Başlarken
Farklı veri kaynaklarından gelen veriler Spark Streaming tarafından işlenir ve sonuçlar harici bir dosya sistemine çıkarılır.
Özellikleri:
- Düşük gecikme süresi
- Hatalardan verimli bir şekilde kurtulabilir
- Yüzlerce düğümde çalışabilir
- Toplu işleme, makine öğrenimi, grafik hesaplama ve diğer çerçevelerle sorunsuz bir şekilde entegre edilebilir.
Yürütme akış şeması:
Spark Streaming, Flink'in aksine gerçek anlamda gerçek zamanlı işleme değil, mikro toplu işleme yapıyor.
Durum: Kelime sıklığı istatistikleri
paket org.apache.spark.examples.streaming org.apache.spark.SparkConf dosyasını içe aktar org.apache.spark.storage.StorageLevel dosyasını içe aktar org.apache.spark.streaming. {Seconds, StreamingContext} dosyasını içe aktarın object NetworkWordCount { def main (args: Array) { eğer (args.length < 2) { System.err.println ("Kullanım: NetworkWordCount < ana bilgisayar adı > < Liman > ") System.exit (1) } StreamingExamples.setStreamingLogLevels () // 1 saniyelik toplu iş boyutuyla bağlamı oluşturun val sparkConf = new SparkConf (). setAppName ("NetworkWordCount") val ssc = new StreamingContext (sparkConf, Seconds (1)) val satırları = ssc.socketTextStream (args (0), args (1) .toInt, StorageLevel.MEMORY_AND_DISK_SER) val sözcükler = lines.flatMap (_. bölme ("")) val wordCounts = word.map (x = > (x, 1)). azaltmaKey (_ + _) wordCounts.print () ssc.start () ssc.awaitTermination () } }
Spark-submit kullanımı:
Öncelikle bir istemci açın ve engellemek için nc -lk 9999 komutunu çalıştırın, veri göndermek için bir süre bekleyin.
Komutu başka bir istemcide yürütün:
spark-submit --master local --class org.apache.spark.examples.streaming.NetworkWordCount --name NetworkWordCount /soft/spark/examples/jars/spark-examples_2.11-2.2.0.jar hadoop09999
Veri göndermek ve kelime sıklığı istatistiklerinin sonuçlarını gözlemlemek için ilk müşteriyi kullanın.
Göndermek için kıvılcım kabuğu kullanın:
Spark Shell istemcisini başlatmak için / soft / spark / bin / spark-shell --master local komutunu çalıştırın.
Ardından istemcide komutu girin:
org.apache.spark.streaming. {Seconds, StreamingContext} dosyasını içe aktarın
val ssc = new StreamingContext (sc, Saniye (1))
val satırları = ssc.socketTextStream ("hadoop0", 9999)
val sözcükler = lines.flatMap (_. bölme (""))
val wordCounts = word.map (x = > (x, 1)). azaltmaKey (_ + _)
wordCounts.print ()
ssc.start ()
ssc.awaitTermination ()
Üretim ortamında test etmek ve kıvılcım göndermek için kıvılcım kabuğu kullanın.
Spark Streaming nasıl çalışır:
Spark Core için çekirdeği RDD'dir. Spark Streaming için çekirdeği DStream'dir. DStream RDD'ye benzer. Esasen bir RDD koleksiyonudur. DStream, veri akışlarını saniye sayısına göre gruplar halinde bölebilir (val ssc = new StreamingContext (sc, Seconds (1)), bu program zaman aralığını ayarlamak için kullanılır). İlk olarak, akış verilerini aldıktan sonra, birden fazla partiye bölünür ve ardından hesaplama için Spark kümesine gönderilir ve son olarak sonuçlar, HDFS'ye veya veri tabanına ve ön uç sayfa görüntüsüne, vb. Gruplar halinde çıkarılır.
Spark Akışı DStream
1. Spark Streaming, İngilizce'de Ayrık Akım olarak adlandırılan ve sürekli bir veri akışını temsil eden Çince'de "Ayrık Akış" a çevrilen DStream adlı üst düzey bir soyutlama sağlar. DStream, Kafka, Flume, ZMQ ve Kinesis gibi giriş veri kaynakları tarafından oluşturulabilir; ayrıca eşleme, azaltma, birleştirme ve pencere gibi diğer DStream'lere daha yüksek dereceli işlevler uygulanarak da oluşturulabilir.
2. DStream'in içinde, aslında sürekli olarak üretilen bir dizi RDD vardır. RDD, Spark Core'un, yani değişmez, dağıtılmış veri kümelerinin temel soyutlamasıdır. DStream'deki her RDD, belirli bir süre için veri içerir.
3. Harita gibi DStream'e uygulanan operatörler, alt katmandaki DStream'deki her RDD'deki işlemlere dönüştürülür. Örneğin, bir DStream üzerinde bir harita işlemi gerçekleştirmek yeni bir DStream oluşturacaktır. Bununla birlikte, en alt seviyede, ilke, harita işleminin giriş DStream'deki her zaman periyodunun RDD'sine uygulanması ve daha sonra yeni RDD'nin, yeni DStream'de o zaman periyodunun bir RDD'si olarak üretilmesidir. Temeldeki RDD'nin dönüştürme işlemi.
4. Halen Spark Core'un hesaplama motoru tarafından uygulanmaktadır. Spark Streaming, Spark Core'u kapsüller, ayrıntıları gizler ve geliştiricilere kullanışlı ve kullanımı kolay bir üst düzey API sağlar.
