Spark修炼之道(进阶篇)——Spark入门到精通:第十六节 Spark Streaming与Kafka-阿里云开发者社区

开发者社区> 周志湖> 正文

Spark修炼之道(进阶篇)——Spark入门到精通:第十六节 Spark Streaming与Kafka

简介: 作者:周志湖 主要内容 Spark Streaming与Kafka版的WordCount示例(一) Spark Streaming与Kafka版的WordCount示例(二) 1. Spark Streaming与Kafka版本的WordCount示例 (一) 启动kafka集群 root@sparkslave02:/hadoopLearning/kafka_2
+关注继续查看

作者:周志湖

主要内容

  1. Spark Streaming与Kafka版的WordCount示例(一)
  2. Spark Streaming与Kafka版的WordCount示例(二)

1. Spark Streaming与Kafka版本的WordCount示例 (一)

  1. 启动kafka集群
root@sparkslave02:/hadoopLearning/kafka_2.10-0.8.2.1# bin/kafka-server-start.sh config/server.properties 
root@sparkslave01:/hadoopLearning/kafka_2.10-0.8.2.1# bin/kafka-server-start.sh config/server.properties 
root@sparkmaster:/hadoopLearning/kafka_2.10-0.8.2.1# bin/kafka-server-start.sh config/server.properties 

向kafka集群发送消息

root@sparkslave01:/hadoopLearning/kafka_2.10-0.8.2.1# bin/kafka-console-producer.sh --broker-list sparkslave01:9092 --sync --topic kafkatopictest
  1. 编写如下程序
import org.apache.kafka.clients.producer.{ProducerConfig, KafkaProducer, ProducerRecord}
import org.apache.log4j.{Level, Logger}

import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.{Logging, SparkConf}

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

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

    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
    //创建ReceiverInputDStream
    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()
  }
}

配置运行参数:
这里写图片描述
具体如下:

sparkmaster:2181  test-consumer-group kafkatopictest 1

sparkmaster:2181,zookeeper监听地址
test-consumer-group, consumer-group的名称,必须和$KAFKA_HOME/config/consumer.properties中的group.id的配置内容一致
kafkatopictest,topic名称
1,线程数

运行KafkaWordCount 后,在producer中输入下列内容

root@sparkslave01:/hadoopLearning/kafka_2.10-0.8.2.1# bin/kafka-console-producer.sh --broker-list sparkslave01:9092 --sync --topic kafkatopictest
[2015-11-04 03:25:39,666] WARN Property topic is not valid (kafka.utils.VerifiableProperties)
Spark
Spark TEST
TEST Spark Streaming

这里写图片描述

得到结果如下:
这里写图片描述

2. Spark Streaming与Kafka版本的WordCount示例(二)

前面的例子中,producer是通过kafka的脚本生成的,本例中将给出通过编写程序生成的producer

// 随机生成1-100间的数字
object KafkaWordCountProducer {

  def main(args: Array[String]) {
    if (args.length < 4) {
      System.err.println("Usage: KafkaWordCountProducer <metadataBrokerList> <topic> " +
        "<messagesPerSec> <wordsPerMessage>")
      System.exit(1)
    }

    val Array(brokers, topic, messagesPerSec, wordsPerMessage) = args

    // Zookeeper连接属性配置
    val props = new HashMap[String, Object]()
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")
    //创建KafkaProducer
    val producer = new KafkaProducer[String, String](props)

    // 向kafka集群发送消息
    while(true) {
      (1 to messagesPerSec.toInt).foreach { messageNum =>
        val str = (1 to wordsPerMessage.toInt).map(x => scala.util.Random.nextInt(10).toString)
          .mkString(" ")

        val message = new ProducerRecord[String, String](topic, null, str)
        producer.send(message)
      }

      Thread.sleep(1000)
    }
  }

}

KafkaWordCountProducer 运行参数设置如下:

sparkmaster:9092 kafkatopictest 5 8

sparkmaster:9092,broker-list
kafkatopictest,top名称
5表示每秒发多少条消息
8表示每条消息中有几个单词

先KafkaWordCountProducer,然后再运行KafkaWordCount ,得到的计算结果如下:
这里写图片描述

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
阿里云服务器端口号设置
阿里云服务器初级使用者可能面临的问题之一. 使用tomcat或者其他服务器软件设置端口号后,比如 一些不是默认的, mysql的 3306, mssql的1433,有时候打不开网页, 原因是没有在ecs安全组去设置这个端口号. 解决: 点击ecs下网络和安全下的安全组 在弹出的安全组中,如果没有就新建安全组,然后点击配置规则 最后如上图点击添加...或快速创建.   have fun!  将编程看作是一门艺术,而不单单是个技术。
3976 0
Spark修炼之道(进阶篇)——Spark入门到精通:第十节 Spark SQL案例实战(一)
作者:周志湖 放假了,终于能抽出时间更新博客了……. 1. 获取数据 本文通过将github上的Spark项目git日志作为数据,对SparkSQL的内容进行详细介绍 数据获取命令如下: [root@master spark]# git log --pretty=format:'{"commit":"%H","author":"%an","author_e
4180 0
列表的简介 | Python从入门到精通:进阶篇之一
列表(List)是一种用于保存一系列有序项目的集合。类似于栈,项目的列表用[]括起来,创建的列表,可以添加、移除、搜索列表中的项目。
1131 0
艾伟_转载:Visual Studio调试之断点进阶篇
文章导航 Visual Studio调试之断点基础篇 Visual Studio调试之断点进阶篇 Visual Studio调试之断点技巧篇 在上一篇文章Visual Studio调试之断点基础篇里面介绍了什么是断点,INT 是Intel系列CPU的一个指令,可以让程序产生一个中断或者异常。
907 0
Spark修炼之道(进阶篇)——Spark入门到精通:第一节 Spark 1.5.0集群搭建
作者:周志湖 网名:摇摆少年梦 微信号:zhouzhihubeyond 本节主要内容 操作系统环境准备 Hadoop 2.4.1集群搭建 Spark 1.5.0 集群部署 注:在利用CentOS 6.5操作系统安装spark 1.5集群过程中,本人发现Hadoop 2.4.1集群可以顺利搭建,但在Spark 1.5.0集群启动时出现了问题(可能原因是64位操作
4548 0
Spark修炼之道(进阶篇)——Spark入门到精通:第十六节 Spark Streaming与Kafka
作者:周志湖 主要内容 Spark Streaming与Kafka版的WordCount示例(一) Spark Streaming与Kafka版的WordCount示例(二) 1. Spark Streaming与Kafka版本的WordCount示例 (一) 启动kafka集群 root@sparkslave02:/hadoopLearning/kafka_2
3827 0
Python 进阶_OOP 面向对象编程_self 的实例绑定
目录 目录 self 和绑定 调用非绑定的方法 self 和绑定 在 Python 中 self 变量是特殊的, 其用于在实例方法中引用该方法所绑定的实例, 换句话说就是 Python 在实例化对象时会自动的将实例对象传递到类体的 self 变量中, 类体中的 self 变量代表了该类的实例对象.
883 0
+关注
99
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
文娱运维技术
立即下载
《SaaS模式云原生数据仓库应用场景实践》
立即下载
《看见新力量:二》电子书
立即下载