Apache Kafka - Intégration avec Spark

Dans ce chapitre, nous discuterons de la manière d'intégrer Apache Kafka avec l'API Spark Streaming.

À propos de Spark

L'API Spark Streaming permet un traitement de flux évolutif, à haut débit et tolérant aux pannes des flux de données en direct. Les données peuvent être ingérées à partir de nombreuses sources telles que Kafka, Flume, Twitter, etc., et peuvent être traitées à l'aide d'algorithmes complexes tels que des fonctions de haut niveau telles que mapper, réduire, joindre et fenêtre. Enfin, les données traitées peuvent être transférées vers des systèmes de fichiers, des bases de données et des tableaux de bord en direct. Les ensembles de données distribués résilients (RDD) sont une structure de données fondamentale de Spark. C'est une collection d'objets distribués immuable. Chaque ensemble de données dans RDD est divisé en partitions logiques, qui peuvent être calculées sur différents nœuds du cluster.

Intégration avec Spark

Kafka est une plate-forme potentielle de messagerie et d'intégration pour le streaming Spark. Kafka sert de hub central pour les flux de données en temps réel et sont traités à l'aide d'algorithmes complexes dans Spark Streaming. Une fois les données traitées, Spark Streaming pourrait publier les résultats dans un autre sujet Kafka ou stocker dans HDFS, des bases de données ou des tableaux de bord. Le diagramme suivant illustre le flux conceptuel.

Maintenant, passons en revue les API Kafka-Spark en détail.

API SparkConf

Il représente la configuration d'une application Spark. Utilisé pour définir divers paramètres Spark sous forme de paires clé-valeur.

La classe SparkConf a les méthodes suivantes -

set(string key, string value) - définir la variable de configuration.
remove(string key) - retirer la clé de la configuration.
setAppName(string name) - définissez le nom de l'application pour votre application.
get(string key) - obtenir la clé

API StreamingContext

Il s'agit du principal point d'entrée de la fonctionnalité Spark. Un SparkContext représente la connexion à un cluster Spark et peut être utilisé pour créer des RDD, des accumulateurs et des variables de diffusion sur le cluster. La signature est définie comme indiqué ci-dessous.

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

master - URL du cluster auquel se connecter (par exemple, mesos: // hôte: port, spark: // hôte: port, local [4]).
appName - un nom pour votre travail, à afficher sur l'interface utilisateur Web du cluster
batchDuration - l'intervalle de temps auquel les données de streaming seront divisées en lots

public StreamingContext(SparkConf conf, Duration batchDuration)

Créez un StreamingContext en fournissant la configuration nécessaire pour un nouveau SparkContext.

conf - Paramètres Spark
batchDuration - l'intervalle de temps auquel les données de streaming seront divisées en lots

API KafkaUtils

L'API KafkaUtils est utilisée pour connecter le cluster Kafka au streaming Spark. Cette API a la signature createStream de la méthode significative définie comme ci-dessous.

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

La méthode illustrée ci-dessus est utilisée pour créer un flux d'entrée qui extrait les messages de Kafka Brokers.

ssc - Objet StreamingContext.
zkQuorum - Collège des gardiens de zoo.
groupId - L'identifiant de groupe pour ce consommateur.
topics - renvoyer une carte des sujets à consommer.
storageLevel - Niveau de stockage à utiliser pour stocker les objets reçus.

L'API KafkaUtils a une autre méthode createDirectStream, qui est utilisée pour créer un flux d'entrée qui extrait directement les messages de Kafka Brokers sans utiliser de récepteur. Ce flux peut garantir que chaque message de Kafka est inclus dans les transformations exactement une fois.

L'exemple d'application est réalisé dans Scala. Pour compiler l'application, veuillez télécharger et installer sbt , scala build tool (similaire à maven). Le code principal de l'application est présenté ci-dessous.

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()
   }
}

Créer un script

L'intégration spark-kafka dépend du jar d'intégration Spark, Spark Streaming et Spark Kafka. Créez un nouveau fichier build.sbt et spécifiez les détails de l'application et sa dépendance. Le sbt téléchargera le fichier jar nécessaire lors de la compilation et de l'emballage de l'application.

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"

Compilation / Emballage

Exécutez la commande suivante pour compiler et empaqueter le fichier jar de l'application. Nous devons soumettre le fichier jar dans la console Spark pour exécuter l'application.

sbt package

Soumettre à Spark

Démarrez l'interface de ligne de commande de Kafka Producer (expliqué dans le chapitre précédent), créez un nouveau sujet appelé my-first-topic et fournissez des exemples de messages comme indiqué ci-dessous.

Another spark test message

Exécutez la commande suivante pour soumettre l'application à la console 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>

L'exemple de sortie de cette application est illustré ci-dessous.

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