Kafka快速入门_卡夫卡教程

S.No

Components and Description

1

Topics

属于特定类别的消息流被称为主题。数据存储在主题中。

主题分为多个分区。对于每个主题，卡夫卡都保留一个分区的最小范围。每个这样的分区都以不可变的有序顺序包含消息。分区被实现为一组相同大小的段文件。

2

Partition

主题可能有很多分区，所以它可以处理任意数量的数据。

3

Partition offset

每个分区消息都有一个称为偏移量的唯一序列标识。

4

Replicas of partition

副本只是分区的备份。副本从不读取或写入数据。它们用于防止数据丢失。

5

Brokers

经纪人是简单的系统，负责维护公布的数据。每个代理可能每个主题有零个或多个分区。假设，如果一个主题和N个代理中有N个分区，则每个代理将有一个分区。

假设某个主题中有N个分区并且N个代理（n + m）多于N个，则第一个N代理将拥有一个分区，下一个M代理将不会拥有该特定主题的任何分区。

假设某个主题中有N个分区，且N个代理的数量少于N个（nm），则每个代理将拥有一个或多个分区共享。由于代理人之间的负载分配不均衡，不推荐这种情况。

6

Kafka Cluster

卡夫卡拥有多个经纪人称为卡夫卡集群。Kafka集群可以在无需停机的情况下进行扩展。这些集群用于管理消息数据的持久性和复制。

7

Producers

制片人是一个或多个卡夫卡主题的发布者。制片人向卡夫卡经纪人发送数据。每当生产者向经纪人发布消息时，经纪人只需将消息附加到最后一个段文件。实际上，该消息将被附加到分区。生产者也可以将消息发送到他们选择的分区。

8

Consumers

消费者从经纪商那里读取数据。消费者通过从代理商处获取数据来订阅一个或多个主题并消费发布的消息。

9

Leader

Leader是负责所有分区读写的节点。每个分区都有一台服务器充当领导者。

10

Follower

遵循领导指示的节点称为追随者。如果领导失败，其中一个追随者将自动成为新领导。追随者扮演正常的消费者角色，拉动消息并更新自己的数据存储。

S.No

Components and Description

1

Broker

Kafka集群通常由多个代理组成，以保持负载平衡。卡夫卡经纪人是无国籍的，所以他们使用ZooKeeper维护他们的集群状态。一个Kafka代理实例可以处理每秒数十万次的读写操作，每个Broker都可以处理TB消息，而不会影响性能。卡夫卡经纪人领导人选举可以由ZooKeeper完成。

2

ZooKeeper

ZooKeeper用于管理和协调Kafka经纪人。ZooKeeper服务主要用于通知生产者和消费者有关Kafka系统中任何新经纪人的存在或Kafka系统中经纪人的失败。根据Zookeeper收到的有关经纪人存在或失败的通知，生产者和消费者就会做出决定，并开始与其他经纪人协调他们的任务。

3

Producers

生产者将数据推送给经纪商。新代理启动后，所有生产者都会搜索它并自动向该新代理发送消息。卡夫卡制片人不会等待经纪人的确认，并且可以像经纪人能够处理的那样快地发送消息。

4

Consumers

由于Kafka代理是无状态的，这意味着消费者必须通过使用分区偏移量来维护消耗了多少消息。如果消费者确认特定的消息偏移量，则意味着消费者已经消费了所有先前的消息。消费者向代理发出一个异步拉取请求，以准备消耗字节缓冲区。消费者可以简单地通过提供偏移值来倒回或跳到分区中的任何点。消费者补偿值由ZooKeeper通知。

$ cd opt/

$ tar -zxf kafka_2.11.0.9.0.0 tar.gz

$ cd kafka_2.11.0.9.0.0

$ bin/kafka-server-start.sh config/server.properties

$ bin/kafka-server-start.sh config/server.properties

[2016-01-02 15:37:30,410] INFO KafkaConfig values:

request.timeout.ms = 30000

log.roll.hours = 168

inter.broker.protocol.version = 0.9.0.X

log.preallocate = false

security.inter.broker.protocol = PLAINTEXT

…………………………………………….

…………………………………………….

$ bin/kafka-server-stop.sh config/server.properties

bin/zookeeper-server-start.sh config/zookeeper.properties

bin/kafka-server-start.sh config/server.properties

821 QuorumPeerMain

928 Kafka

931 Jps

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1

--partitions 1 --topic topic-name

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1   

--partitions 1 --topic Hello-Kafka

bin/kafka-topics.sh --list --zookeeper localhost:2181

Hello-Kafka

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-name

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka

$ bin/kafka-console-producer.sh --broker-list localhost:9092

--topic Hello-Kafka[2016-01-16 13:50:45,931]

WARN property topic is not valid (kafka.utils.Verifia-bleProperties)

Hello

My first message

My second message

bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic topic-name

--from-beginning

bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Hello-Kafka

--from-beginning

Hello

My first message

My second message

# The id of the broker. This must be set to a unique integer for each broker.

broker.id=1

# The port the socket server listens on

port=9093

# A comma seperated list of directories under which to store log files

log.dirs=/tmp/kafka-logs-1

# The id of the broker. This must be set to a unique integer for each broker.

broker.id=2

# The port the socket server listens on

port=9094

# A comma seperated list of directories under which to store log files

log.dirs=/tmp/kafka-logs-2

Broker1

bin/kafka-server-start.sh config/server.properties

Broker2

bin/kafka-server-start.sh config/server-one.properties

Broker3

bin/kafka-server-start.sh config/server-two.properties

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3

-partitions 1 --topic topic-name

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3

-partitions 1 --topic Multibrokerapplication

created topic “Multibrokerapplication”

bin/kafka-topics.sh --describe --zookeeper localhost:2181

--topic Multibrokerappli-cation

bin/kafka-topics.sh --describe --zookeeper localhost:2181

--topic Multibrokerappli-cation

Topic:Multibrokerapplication    PartitionCount:1

ReplicationFactor:3 Configs:

Topic:Multibrokerapplication Partition:0 Leader:0

Replicas:0,2,1 Isr:0,2,1

bin/kafka-console-producer.sh --broker-list localhost:9092

--topic Multibrokerapplication

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Multibrokerapplication

[2016-01-20 19:27:21,045] WARN Property topic is not valid (kafka.utils.Verifia-bleProperties)

This is single node-multi broker demo

This is the second message

bin/kafka-console-consumer.sh --zookeeper localhost:2181

—topic Multibrokerapplica-tion --from-beginning

bin/kafka-console-consumer.sh --zookeeper localhost:2181

—topic Multibrokerapplica-tion —from-beginning

This is single node-multi broker demo

This is the second message

bin/kafka-topics.sh —zookeeper localhost:2181 --alter --topic topic_name

--parti-tions count

We have already created a topic “Hello-Kafka” with single partition count and one replica factor.

Now using “alter” command we have changed the partition count.

bin/kafka-topics.sh --zookeeper localhost:2181

--alter --topic Hello-kafka --parti-tions 2

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic topic_name

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic Hello-kafka

> Topic Hello-kafka marked for deletion

producer.send(new ProducerRecord<byte[],byte[]>(topic,

partition, key1, value1) , callback);

public void flush()

public Map metrics()

public void send(KeyedMessaget<k,v> message)

- sends the data to a single topic,par-titioned by key using either sync or async producer.

public void send(List<KeyedMessage<k,v>>messages)

- sends data to multiple topics.

Properties prop = new Properties();

prop.put(producer.type,”async”)

ProducerConfig config = new ProducerConfig(prop);

S.No

Configuration Settings and Description

1

client.id

确定生产者申请

2

producer.type

无论是同步还是异步

3

acks

acks配置控制生产者请求下的标准被认为是完整的。

4

retries

如果生产者请求失败，然后自动重试具有特定的值。

5

bootstrap.servers

引导经纪人名单。

6

linger.ms

如果你想减少请求的数量，你可以将linger.ms设置为大于某个值的值。

7

key.serializer

串行器接口的关键。

8

value.serializer

串行器接口的值。

9

batch.size

缓冲区大小。

10

buffer.memory

控制制造商可用于缓冲的总内存量。

public ProducerRecord (string topic, int partition, k key, v value)

public ProducerRecord (string topic, k key, v value)

public ProducerRecord (string topic, v value)

S.No

类方法和描述

1

public string topic()

主题将附加到记录。

2

public K key()

将包含在记录中的密钥。如果没有这样的密钥，null将在这里重新转向。

3

public V value()

记录内容。

4

partition()

记录的分区计数

//import util.properties packages

import java.util.Properties;

//import simple producer packages

import org.apache.kafka.clients.producer.Producer;

//import KafkaProducer packages

import org.apache.kafka.clients.producer.KafkaProducer;

//import ProducerRecord packages

import org.apache.kafka.clients.producer.ProducerRecord;

//Create java class named “SimpleProducer”

public class SimpleProducer {

   public static void main(String[] args) throws Exception{

      // Check arguments length value

      if(args.length == 0){

         System.out.println("Enter topic name”);

         return;

      }

      //Assign topicName to string variable

      String topicName = args[0].toString();

      // create instance for properties to access producer configs   

      Properties props = new Properties();

      //Assign localhost id

      props.put("bootstrap.servers", “localhost:9092");

      //Set acknowledgements for producer requests.      

      props.put("acks", “all");

      //If the request fails, the producer can automatically retry,

      props.put("retries", 0);

      //Specify buffer size in config

      props.put("batch.size", 16384);

      //Reduce the no of requests less than 0   

      props.put("linger.ms", 1);

      //The buffer.memory controls the total amount of memory available to the producer for buffering.   

      props.put("buffer.memory", 33554432);

      props.put("key.serializer",

         "org.apache.kafka.common.serializa-tion.StringSerializer");

      props.put("value.serializer",

         "org.apache.kafka.common.serializa-tion.StringSerializer");

      Producer<String, String> producer = new KafkaProducer

         <String, String>(props);

      for(int i = 0; i < 10; i++)

         producer.send(new ProducerRecord<String, String>(topicName,

            Integer.toString(i), Integer.toString(i)));

               System.out.println(“Message sent successfully”);

               producer.close();

   }

}

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleProducer <topic-name>

Message sent successfully

To check the above output open new terminal and type Consumer CLI command to receive messages.

>> bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic <topic-name> —from-beginning

1

2

3

4

5

6

7

8

9

10

public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs)

S.No

Method and Description

1

public java.util.Set<TopicPar-tition> assignment()

获取消费者当前分配的一组分区。

2

public string subscription()

订阅给定的主题列表以获取动态签署的分区。

3

public void sub-scribe(java.util.List<java.lang.String> topics, ConsumerRe-balanceListener listener)

订阅给定的主题列表以获取动态签署的分区。

4

public void unsubscribe()

取消订阅给定分区列表中的主题。

5

public void sub-scribe(java.util.List<java.lang.String> topics)

订阅给定的主题列表以获取动态签署的分区。如果给定的主题列表为空，则将其视为与取消订阅（）相同。

6

public void sub-scribe(java.util.regex.Pattern pattern, ConsumerRebalanceLis-tener listener)

参数模式以正则表达式的格式引用订阅模式，而侦听器参数从订阅模式中获取通知。

7

public void as-sign(java.util.List<TopicParti-tion> partitions)

手动将分区列表分配给客户。

8

poll()

获取使用其中一个订阅/分配API指定的主题或分区的数据。如果在轮询数据之前没有订阅主题，这将返回错误。

9

public void commitSync()

提交最后一次poll（）返回的主题和分区的所有sub-scribed列表的偏移量。同样的操作应用于commitAsyn（）。

10

public void seek(TopicPartition partition, long offset)

获取消费者将在下一个poll（）方法中使用的当前偏移值。

11

public void resume()

恢复暂停的分区。

12

public void wakeup()

唤醒消费者。

public ConsumerRecord(string topic,int partition, long offset,K key, V value)

public ConsumerRecords(java.util.Map<TopicPartition,java.util.List

<Consumer-Record>K,V>>> records)

S.No

方法和描述

1

public int count（）

所有主题的记录数。

2

public Set partitions()

该记录集中包含数据的分区集（如果没有数据返回，则该集为空）。

3

public Iterator iterator()

迭代器使您能够遍历集合，获取或重新移动元素。

4

public List records()

获取给定分区的记录列表。

S.No

Settings and Description

1

bootstrap.servers

引导经纪人名单。

2

group.id

将个人消费者分配给组。

3

enable.auto.commit

如果值为true，则为偏移启用自动提交，否则不提交。

4

auto.commit.interval.ms

返回更新的消耗偏移量被写入ZooKeeper的频率。

5

session.timeout.ms

指示在放弃并继续使用消息之前，Kafka将等待多少毫秒以等待ZooKeeper响应请求（读取或写入）。

import java.util.Properties;

import java.util.Arrays;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.apache.kafka.clients.consumer.ConsumerRecords;

import org.apache.kafka.clients.consumer.ConsumerRecord;

public class SimpleConsumer {

   public static void main(String[] args) throws Exception {

      if(args.length == 0){

         System.out.println("Enter topic name");

         return;

      }

      //Kafka consumer configuration settings

      String topicName = args[0].toString();

      Properties props = new Properties();

      props.put("bootstrap.servers", "localhost:9092");

      props.put("group.id", "test");

      props.put("enable.auto.commit", "true");

      props.put("auto.commit.interval.ms", "1000");

      props.put("session.timeout.ms", "30000");

      props.put("key.deserializer",

         "org.apache.kafka.common.serializa-tion.StringDeserializer");

      props.put("value.deserializer",

         "org.apache.kafka.common.serializa-tion.StringDeserializer");

      KafkaConsumer<String, String> consumer = new KafkaConsumer

         <String, String>(props);

      //Kafka Consumer subscribes list of topics here.

      consumer.subscribe(Arrays.asList(topicName))

      //print the topic name

      System.out.println("Subscribed to topic " + topicName);

      int i = 0;

      while (true) {

         ConsumerRecords<String, String> records = con-sumer.poll(100);

         for (ConsumerRecord<String, String> record : records)

         // print the offset,key and value for the consumer records.

         System.out.printf("offset = %d, key = %s, value = %s\n",

            record.offset(), record.key(), record.value());

      }

   }

}

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleConsumer <topic-name>

Subscribed to topic Hello-Kafka

offset = 3, key = null, value = Hello Consumer

import java.util.Properties;

import java.util.Arrays;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.apache.kafka.clients.consumer.ConsumerRecords;

import org.apache.kafka.clients.consumer.ConsumerRecord;

public class ConsumerGroup {

   public static void main(String[] args) throws Exception {

      if(args.length < 2){

         System.out.println("Usage: consumer <topic> <groupname>");

         return;

      }

      String topic = args[0].toString();

      String group = args[1].toString();

      Properties props = new Properties();

      props.put("bootstrap.servers", "localhost:9092");

      props.put("group.id", group);

      props.put("enable.auto.commit", "true");

      props.put("auto.commit.interval.ms", "1000");

      props.put("session.timeout.ms", "30000");

      props.put("key.deserializer",          

         "org.apache.kafka.common.serializa-tion.StringDeserializer");

      props.put("value.deserializer",

         "org.apache.kafka.common.serializa-tion.StringDeserializer");

      KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

      consumer.subscribe(Arrays.asList(topic));

      System.out.println("Subscribed to topic " + topic);

      int i = 0;

      while (true) {

         ConsumerRecords<String, String> records = con-sumer.poll(100);

            for (ConsumerRecord<String, String> record : records)

               System.out.printf("offset = %d, key = %s, value = %s\n",

               record.offset(), record.key(), record.value());

      }     

   }  

}

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*" ConsumerGroup.java

>>java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*":.

ConsumerGroup <topic-name> my-group

>>java -cp "/home/bala/Workspace/kafka/kafka_2.11-0.9.0.0/libs/*":.

ConsumerGroup <topic-name> my-group

Test consumer group 01

Test consumer group 02

Subscribed to topic Hello-kafka

offset = 3, key = null, value = Test consumer group 01

Subscribed to topic Hello-kafka

offset = 3, key = null, value = Test consumer group 02

bin/kafka-run-class.sh package.class - - options

public ZkHosts(String brokerZkStr, String brokerZkPath)

public ZkHosts(String brokerZkStr)

public KafkaConfig(BrokerHosts hosts, string topic)

public SpoutConfig(BrokerHosts hosts, string topic, string zkRoot, string id)

public SchemeAsMultiScheme(Scheme scheme)

// ZooKeeper connection string

BrokerHosts hosts = new ZkHosts(zkConnString);

//Creating SpoutConfig Object

SpoutConfig spoutConfig = new SpoutConfig(hosts,

   topicName, "/" + topicName UUID.randomUUID().toString());

//convert the ByteBuffer to String.

spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

//Assign SpoutConfig to KafkaSpout.

KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);

import java.util.Map;

import backtype.storm.tuple.Tuple;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Values;

import backtype.storm.task.OutputCollector;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.IRichBolt;

import backtype.storm.task.TopologyContext;

public class SplitBolt implements IRichBolt {

   private OutputCollector collector;

   @Override

   public void prepare(Map stormConf, TopologyContext context,

      OutputCollector collector) {

      this.collector = collector;

   }

   @Override

   public void execute(Tuple input) {

      String sentence = input.getString(0);

      String[] words = sentence.split(" ");

      for(String word: words) {

         word = word.trim();

         if(!word.isEmpty()) {

            word = word.toLowerCase();

            collector.emit(new Values(word));

         }

      }

      collector.ack(input);

   }

   @Override

   public void declareOutputFields(OutputFieldsDeclarer declarer) {

      declarer.declare(new Fields("word"));

   }

   @Override

   public void cleanup() {}

   @Override

   public Map<String, Object> getComponentConfiguration() {

      return null;

   }

}

import java.util.Map;

import java.util.HashMap;

import backtype.storm.tuple.Tuple;

import backtype.storm.task.OutputCollector;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.IRichBolt;

import backtype.storm.task.TopologyContext;

public class CountBolt implements IRichBolt{

   Map<String, Integer> counters;

   private OutputCollector collector;

   @Override

   public void prepare(Map stormConf, TopologyContext context,

   OutputCollector collector) {

      this.counters = new HashMap<String, Integer>();

      this.collector = collector;

   }

   @Override

   public void execute(Tuple input) {

      String str = input.getString(0);

      if(!counters.containsKey(str)){

         counters.put(str, 1);

      }else {

         Integer c = counters.get(str) +1;

         counters.put(str, c);

      }

      collector.ack(input);

   }

   @Override

   public void cleanup() {

      for(Map.Entry<String, Integer> entry:counters.entrySet()){

         System.out.println(entry.getKey()+" : " + entry.getValue());

      }

   }

   @Override

   public void declareOutputFields(OutputFieldsDeclarer declarer) {

   }

   @Override

   public Map<String, Object> getComponentConfiguration() {

      return null;

   }

}

import backtype.storm.Config;

import backtype.storm.LocalCluster;

import backtype.storm.topology.TopologyBuilder;

import java.util.ArrayList;

import java.util.List;

import java.util.UUID;

import backtype.storm.spout.SchemeAsMultiScheme;

import storm.kafka.trident.GlobalPartitionInformation;

import storm.kafka.ZkHosts;

import storm.kafka.Broker;

import storm.kafka.StaticHosts;

import storm.kafka.BrokerHosts;

import storm.kafka.SpoutConfig;

import storm.kafka.KafkaConfig;

import storm.kafka.KafkaSpout;

import storm.kafka.StringScheme;

public class KafkaStormSample {

   public static void main(String[] args) throws Exception{

      Config config = new Config();

      config.setDebug(true);

      config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);

      String zkConnString = "localhost:2181";

      String topic = "my-first-topic";

      BrokerHosts hosts = new ZkHosts(zkConnString);

      SpoutConfig kafkaSpoutConfig = new SpoutConfig (hosts, topic, "/" + topic,    

         UUID.randomUUID().toString());

      kafkaSpoutConfig.bufferSizeBytes = 1024 * 1024 * 4;

      kafkaSpoutConfig.fetchSizeBytes = 1024 * 1024 * 4;

      kafkaSpoutConfig.forceFromStart = true;

      kafkaSpoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

      TopologyBuilder builder = new TopologyBuilder();

      builder.setSpout("kafka-spout", new KafkaSpout(kafkaSpoutCon-fig));

      builder.setBolt("word-spitter", new SplitBolt()).shuffleGroup-ing("kafka-spout");

      builder.setBolt("word-counter", new CountBolt()).shuffleGroup-ing("word-spitter");

      LocalCluster cluster = new LocalCluster();

      cluster.submitTopology("KafkaStormSample", config, builder.create-Topology());

      Thread.sleep(10000);

      cluster.shutdown();

   }

}

javac -cp "/path/to/Kafka/apache-storm-0.9.5/lib/*" *.java

hello

kafka

storm

spark

test message

another test message

java -cp“/path/to/Kafka/apache-storm-0.9.5/lib/*”:. KafkaStormSample

storm : 1

test : 2

spark : 1

another : 1

kafka : 1

hello : 1

message : 2

public StreamingContext(String master, String appName, Duration batchDuration,

   String sparkHome, scala.collection.Seq<String> jars,

   scala.collection.Map<String,String> environment)

public StreamingContext(SparkConf conf, Duration batchDuration)

public static ReceiverInputDStream<scala.Tuple2<String,String>> createStream(

   StreamingContext ssc, String zkQuorum, String groupId,

   scala.collection.immutable.Map<String,Object> topics, StorageLevel storageLevel)

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

   }

}

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"

sbt package

Another spark test message

/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 ..