Apache Kafka - Integração com Spark

Neste capítulo, discutiremos sobre como integrar o Apache Kafka com a API Spark Streaming.

Sobre o Spark

A API Spark Streaming permite o processamento escalonável, de alto rendimento e tolerante a falhas de fluxos de dados ao vivo. Os dados podem ser ingeridos de muitas fontes, como Kafka, Flume, Twitter, etc., e podem ser processados usando algoritmos complexos, como funções de alto nível, como mapear, reduzir, juntar e janela. Finalmente, os dados processados podem ser enviados para sistemas de arquivos, bancos de dados e painéis dinâmicos. Resilient Distributed Datasets (RDD) é uma estrutura de dados fundamental do Spark. É uma coleção imutável de objetos distribuídos. Cada conjunto de dados em RDD é dividido em partições lógicas, que podem ser calculadas em diferentes nós do cluster.

Integração com Spark

Kafka é uma plataforma potencial de mensagens e integração para streaming do Spark. O Kafka atua como o hub central para fluxos de dados em tempo real e são processados usando algoritmos complexos no Spark Streaming. Depois que os dados são processados, o Spark Streaming pode publicar os resultados em outro tópico do Kafka ou armazenar em HDFS, bancos de dados ou painéis. O diagrama a seguir descreve o fluxo conceitual.

Agora, vamos examinar em detalhes a API do Kafka-Spark.

API SparkConf

Ele representa a configuração de um aplicativo Spark. Usado para definir vários parâmetros do Spark como pares de chave-valor.

A classe SparkConf tem os seguintes métodos -

set(string key, string value) - definir variável de configuração.
remove(string key) - remove a chave da configuração.
setAppName(string name) - definir o nome do aplicativo para seu aplicativo.
get(string key) - pegue a chave

API StreamingContext

Este é o principal ponto de entrada para a funcionalidade do Spark. Um SparkContext representa a conexão com um cluster Spark e pode ser usado para criar RDDs, acumuladores e variáveis de transmissão no cluster. A assinatura é definida conforme mostrado abaixo.

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

master - URL do cluster ao qual se conectar (por exemplo, mesos: // host: porta, spark: // host: porta, local [4]).
appName - um nome para seu trabalho, para exibir na IU da web do cluster
batchDuration - o intervalo de tempo em que os dados de streaming serão divididos em lotes

public StreamingContext(SparkConf conf, Duration batchDuration)

Crie um StreamingContext fornecendo a configuração necessária para um novo SparkContext.

conf - Parâmetros do Spark
batchDuration - o intervalo de tempo em que os dados de streaming serão divididos em lotes

API KafkaUtils

A API KafkaUtils é usada para conectar o cluster Kafka ao streaming do Spark. Esta API tem a assinatura do método createStream significativa definida como abaixo.

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

O método mostrado acima é usado para criar um fluxo de entrada que extrai mensagens dos Kafka Brokers.

ssc - Objeto StreamingContext.
zkQuorum - Quorum do Zookeeper.
groupId - O ID do grupo para este consumidor.
topics - retorna um mapa de tópicos para consumir.
storageLevel - Nível de armazenamento a ser usado para armazenar os objetos recebidos.

A API KafkaUtils tem outro método createDirectStream, que é usado para criar um fluxo de entrada que extrai mensagens diretamente dos Kafka Brokers sem usar nenhum receptor. Esse fluxo pode garantir que cada mensagem de Kafka seja incluída nas transformações exatamente uma vez.

O aplicativo de amostra é feito em Scala. Para compilar o aplicativo, baixe e instale o sbt , a ferramenta de construção scala (semelhante ao maven). O código do aplicativo principal é apresentado a seguir.

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

Script de construção

A integração spark-kafka depende do jarro de integração spark, spark streaming e spark Kafka. Crie um novo arquivo build.sbt e especifique os detalhes do aplicativo e sua dependência. O sbt irá baixar o jar necessário enquanto compila e empacota o aplicativo.

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"

Compilação / embalagem

Execute o seguinte comando para compilar e empacotar o arquivo jar do aplicativo. Precisamos enviar o arquivo jar para o console do Spark para executar o aplicativo.

sbt package

Enviando para o Spark

Inicie a CLI do Kafka Producer (explicado no capítulo anterior), crie um novo tópico chamado my-first-topic e forneça algumas mensagens de amostra, conforme mostrado abaixo.

Another spark test message

Execute o seguinte comando para enviar o aplicativo ao console do 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>

O exemplo de saída deste aplicativo é mostrado abaixo.

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