Apache Kafka-Spark와 통합

이 장에서는 Apache Kafka를 Spark Streaming API와 통합하는 방법에 대해 설명합니다.

Spark 정보

Spark Streaming API는 실시간 데이터 스트림의 확장 가능하고 처리량이 높고 내결함성이있는 스트림 처리를 지원합니다. 데이터는 Kafka, Flume, Twitter 등과 같은 여러 소스에서 수집 할 수 있으며 map, reduce, join 및 window와 같은 고급 기능과 같은 복잡한 알고리즘을 사용하여 처리 할 수 있습니다. 마지막으로 처리 된 데이터는 파일 시스템, 데이터베이스 및 라이브 대시 보드로 푸시 될 수 있습니다. RDD (Resilient Distributed Dataset)는 Spark의 기본 데이터 구조입니다. 불변의 분산 된 개체 모음입니다. RDD의 각 데이터 세트는 클러스터의 다른 노드에서 계산 될 수있는 논리 파티션으로 나뉩니다.

Spark와 통합

Kafka는 Spark 스트리밍을위한 잠재적 메시징 및 통합 플랫폼입니다. Kafka는 실시간 데이터 스트림의 중앙 허브 역할을하며 Spark Streaming에서 복잡한 알고리즘을 사용하여 처리됩니다. 데이터가 처리되면 Spark Streaming은 결과를 또 다른 Kafka 주제에 게시하거나 HDFS, 데이터베이스 또는 대시 보드에 저장할 수 있습니다. 다음 다이어그램은 개념적 흐름을 보여줍니다.

이제 Kafka-Spark API를 자세히 살펴 보겠습니다.

SparkConf API

Spark 애플리케이션의 구성을 나타냅니다. 다양한 Spark 매개 변수를 키-값 쌍으로 설정하는 데 사용됩니다.

SparkConf 클래스에는 다음과 같은 메서드가 있습니다.

set(string key, string value) − 구성 변수를 설정합니다.
remove(string key) − 구성에서 키를 제거합니다.
setAppName(string name) − 애플리케이션의 애플리케이션 이름을 설정합니다.
get(string key) − 열쇠 받기

StreamingContext API

이것이 Spark 기능의 주요 진입 점입니다. SparkContext는 Spark 클러스터에 대한 연결을 나타내며 클러스터에서 RDD, 누산기 및 브로드 캐스트 변수를 만드는 데 사용할 수 있습니다. 서명은 아래와 같이 정의됩니다.

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

master − 연결할 클러스터 URL (예 : mesos : // host : port, spark : // host : port, local [4]).
appName − 클러스터 웹 UI에 표시 할 작업 이름
batchDuration − 스트리밍 데이터가 배치로 분할되는 시간 간격

public StreamingContext(SparkConf conf, Duration batchDuration)

새 SparkContext에 필요한 구성을 제공하여 StreamingContext를 만듭니다.

conf − Spark 매개 변수
batchDuration − 스트리밍 데이터가 배치로 분할되는 시간 간격

KafkaUtils API

KafkaUtils API는 Kafka 클러스터를 Spark 스트리밍에 연결하는 데 사용됩니다. 이 API는 아래와 같이 정의 된 signifi-cant 메소드 createStream 시그니처를 가지고 있습니다.

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

위에 표시된 방법은 Kafka Brokers에서 메시지를 가져 오는 입력 스트림을 만드는 데 사용됩니다.

ssc − StreamingContext 객체.
zkQuorum − 사육사 쿼럼.
groupId −이 소비자의 그룹 ID.
topics − 소비 할 주제지도를 반환합니다.
storageLevel − 수신 된 객체를 저장하는 데 사용할 저장 수준.

KafkaUtils API에는 수신자를 사용하지 않고 Kafka Brokers에서 직접 메시지를 가져 오는 입력 스트림을 만드는 데 사용되는 또 다른 메서드 createDirectStream이 있습니다. 이 스트림은 Kafka의 각 메시지가 변환에 정확히 한 번 포함되도록 보장 할 수 있습니다.

샘플 애플리케이션은 Scala에서 수행됩니다. 애플리케이션을 컴파일하려면 sbt , scala 빌드 도구 (maven과 유사) 를 다운로드하여 설치하십시오 . 주요 애플리케이션 코드는 다음과 같습니다.

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

빌드 스크립트

spark-kafka 통합은 spark, spark 스트리밍 및 spark Kafka 통합 jar에 따라 다릅니다. 새 파일 build.sbt를 만들고 응용 프로그램 세부 정보와 해당 종속성을 지정합니다. SBT는 컴파일하고 응용 프로그램을 포장하는 동안 필요한 항아리를 다운로드합니다.

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"

편집 / 패키징

다음 명령을 실행하여 애플리케이션의 jar 파일을 컴파일하고 패키징합니다. 애플리케이션을 실행하려면 jar 파일을 Spark 콘솔에 제출해야합니다.

sbt package

Spark에 제출

Kafka Producer CLI (이전 장에서 설명)를 시작하고 my-first-topic 이라는 새 주제를 만들고 아래와 같이 몇 가지 샘플 메시지를 제공합니다.

Another spark test message

다음 명령을 실행하여 애플리케이션을 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>

이 애플리케이션의 샘플 출력은 다음과 같습니다.

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