Apache Kafka - Integracja ze Spark

W tym rozdziale omówimy sposób integracji Apache Kafka z Spark Streaming API.

O Spark

Spark Streaming API umożliwia skalowalne, wysokoprzepustowe, odporne na błędy przetwarzanie strumieni danych na żywo. Dane mogą być pozyskiwane z wielu źródeł, takich jak Kafka, Flume, Twitter itp., I mogą być przetwarzane przy użyciu złożonych algorytmów, takich jak funkcje wysokiego poziomu, takie jak mapowanie, redukcja, łączenie i okno. Wreszcie przetworzone dane mogą być wypychane do systemów plików, baz danych i działających na żywo tablic rozdzielczych. Resilient Distributed Datasets (RDD) to podstawowa struktura danych platformy Spark. Jest to niezmienny, rozproszony zbiór obiektów. Każdy zbiór danych w RDD jest podzielony na logiczne partycje, które mogą być obliczane na różnych węzłach klastra.

Integracja z Spark

Kafka to potencjalna platforma przesyłania wiadomości i integracji dla przesyłania strumieniowego Spark. Kafka działa jako centralne centrum strumieni danych w czasie rzeczywistym i są przetwarzane przy użyciu złożonych algorytmów w usłudze Spark Streaming. Po przetworzeniu danych Spark Streaming może publikować wyniki w kolejnym temacie Kafki lub przechowywać w HDFS, bazach danych lub pulpitach nawigacyjnych. Poniższy diagram przedstawia koncepcyjny przepływ.

Przejdźmy teraz szczegółowo do interfejsu API Kafka-Spark.

SparkConf API

Reprezentuje konfigurację dla aplikacji Spark. Służy do ustawiania różnych parametrów Spark jako par klucz-wartość.

Klasa SparkConf ma następujące metody -

set(string key, string value) - ustaw zmienną konfiguracyjną.
remove(string key) - usuń klucz z konfiguracji.
setAppName(string name) - ustaw nazwę aplikacji dla swojej aplikacji.
get(string key) - zdobądź klucz

StreamingContext API

To jest główny punkt wejścia do funkcjonalności Spark. SparkContext reprezentuje połączenie z klastrem Spark i może służyć do tworzenia RDD, akumulatorów i zmiennych emisji w klastrze. Podpis jest zdefiniowany w sposób pokazany poniżej.

public StreamingContext(String master, String appName, Duration batchDuration, 
   String sparkHome, scala.collection.Seq<String> jars, 
   scala.collection.Map<String,String> environment)

master - adres URL klastra do połączenia (np. Mesos: // host: port, spark: // host: port, local [4]).
appName - nazwa pracy, która ma być wyświetlana w interfejsie WWW klastra
batchDuration - przedział czasu, w którym dane przesyłane strumieniowo zostaną podzielone na partie

public StreamingContext(SparkConf conf, Duration batchDuration)

Utwórz StreamingContext, podając konfigurację niezbędną dla nowego SparkContext.

conf - Parametry iskry
batchDuration - przedział czasu, w którym dane przesyłane strumieniowo zostaną podzielone na partie

KafkaUtils API

Interfejs API KafkaUtils służy do łączenia klastra Kafka z przesyłaniem strumieniowym Spark. Ten interfejs API ma znaczącą metodę createStream podpisaną poniżej.

public static ReceiverInputDStream<scala.Tuple2<String,String>> createStream(
   StreamingContext ssc, String zkQuorum, String groupId,
   scala.collection.immutable.Map<String,Object> topics, StorageLevel storageLevel)

Przedstawiona powyżej metoda służy do tworzenia strumienia wejściowego, który pobiera wiadomości od brokerów platformy Kafka.

ssc - Obiekt StreamingContext.
zkQuorum - Kworum Zookeeper.
groupId - identyfikator grupy dla tego konsumenta.
topics - zwróć mapę tematów do skonsumowania.
storageLevel - Poziom pamięci do przechowywania odebranych obiektów.

Interfejs API KafkaUtils ma inną metodę createDirectStream, która służy do tworzenia strumienia wejściowego, który bezpośrednio pobiera wiadomości od Brokerów Kafka bez użycia żadnego odbiornika. Ten strumień może zagwarantować, że każda wiadomość od Kafki zostanie uwzględniona w przekształceniach dokładnie raz.

Przykładowa aplikacja jest wykonywana w Scali. Aby skompilować aplikację, pobierz i zainstaluj sbt , narzędzie do kompilacji scala (podobne do mavena). Poniżej przedstawiono główny kod aplikacji.

import java.util.HashMap

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, Produc-erRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._

object KafkaWordCount {
   def main(args: Array[String]) {
      if (args.length < 4) {
         System.err.println("Usage: KafkaWordCount <zkQuorum><group> <topics> <numThreads>")
         System.exit(1)
      }

      val Array(zkQuorum, group, topics, numThreads) = args
      val sparkConf = new SparkConf().setAppName("KafkaWordCount")
      val ssc = new StreamingContext(sparkConf, Seconds(2))
      ssc.checkpoint("checkpoint")

      val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
      val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
      val words = lines.flatMap(_.split(" "))
      val wordCounts = words.map(x => (x, 1L))
         .reduceByKeyAndWindow(_ + _, _ - _, Minutes(10), Seconds(2), 2)
      wordCounts.print()

      ssc.start()
      ssc.awaitTermination()
   }
}

Zbuduj skrypt

Integracja iskra-kafka zależy od iskry, przesyłania strumieniowego iskier i słoika integracji Spark Kafka. Utwórz nowy plik build.sbt i określ szczegóły aplikacji oraz jej zależności. SBT pobierze niezbędne słoik podczas kompilacji i pakowania aplikacji.

name := "Spark Kafka Project"
version := "1.0"
scalaVersion := "2.10.5"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"
libraryDependencies += "org.apache.spark" %% "spark-streaming" % "1.6.0"
libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka" % "1.6.0"

Kompilacja / Pakowanie

Uruchom następujące polecenie, aby skompilować i spakować plik jar aplikacji. Musimy przesłać plik jar do konsoli Spark, aby uruchomić aplikację.

sbt package

Przesyłanie do Spark

Uruchom interfejs wiersza polecenia Kafka Producer (wyjaśniony w poprzednim rozdziale), utwórz nowy temat o nazwie mój-pierwszy-temat i podaj kilka przykładowych komunikatów, jak pokazano poniżej.

Another spark test message

Uruchom następujące polecenie, aby przesłać aplikację do konsoli Spark.

/usr/local/spark/bin/spark-submit --packages org.apache.spark:spark-streaming
-kafka_2.10:1.6.0 --class "KafkaWordCount" --master local[4] target/scala-2.10/spark
-kafka-project_2.10-1.0.jar localhost:2181 <group name> <topic name> <number of threads>

Przykładowe dane wyjściowe tej aplikacji przedstawiono poniżej.

spark console messages ..
(Test,1)
(spark,1)
(another,1)
(message,1)
spark console message ..