注意:清楚flume版本，不同版本配置的參數(shù)值不一樣顾彰，此處是CDH5.7.2對應的flume版本是1.6.0
系統(tǒng)環(huán)境
虛擬機版本: VMware Workstation 10.0
節(jié)點: master (3g內(nèi)存) ,slave1(1g內(nèi)存)
CDH版本: 5.7.2
操作系統(tǒng): Centos 6.8
flume版本: 1.6.0
kafka版本: 0.10.0
spark版本: 1.6.0
master數(shù)據(jù)流向的服務器配置文件
a1.sources= r1
a1.sinks= k1
a1.channels= c1
#Describe/configure the source
a1.sources.r1.type= avro
a1.sources.r1.channels= c1
a1.sources.r1.bind= master
a1.sources.r1.port= 4545
#Describe the sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = sparkstreaming
a1.sinks.k1.brokerList = localhost:9092
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 20
a1.sinks.k1.channel = c1
#Use a channel which buffers events in memory
a1.channels.c1.type= memory
a1.channels.c1.keep-alive= 10
a1.channels.c1.capacity= 100000
a1.channels.c1.transactionCapacity= 100000
slave1數(shù)據(jù)流出的服務器配置文件
a1.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.type = spooldir
a1.sources.r1.channels = c1
a1.sources.r1.spoolDir = /opt/cloudera/parcels/CDH/lib/flume-ng/logs
#a1.sources.r1.fileHeader = true
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = master
a1.sinks.k1.port = 4545
具體步驟
注意，最好每一步都分開一個終端來啟動
啟動這些步驟前需要先啟動zookeeper胃碾，在啟動kafka
#第一步master創(chuàng)建kafka topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic sparkstreaming
#第二步 mater 啟動flume
bin/flume-ng agent --conf /opt/cloudera/parcels/CDH/lib/flume-ng/conf/ -f /opt/cloudera/parcels/CDH/lib/flume-ng/conf/flume-flume-kafka.conf -Dflume.root.logger=INFO,console -n a1
#第三步 slave1啟動flume
bin/flume-ng agent --conf /opt/cloudera/parcels/CDH/lib/flume-ng/conf/ -f /opt/cloudera/parcels/CDH/lib/flume-ng/conf/flume-flume.conf -Dflume.root.logger=INFO,console -n a1
#第四步 slave1使用腳本想目錄增加數(shù)據(jù)
for((i=1;i<=1000;i++));
do
sleep 2;
echo "hello world hello world liujm tljsdkjflsakd hello world hello" >> /opt/cloudera/parcels/CDH/lib/flume-ng/logs/test.log;
done
#第五步master啟動kafka消費者
bin/kafka-console-consumer.sh -zookeeper localhost:2181--from-beginning --topic sparkstreaming
#第六步master啟動spark-streaming(使用spark提供的例子)[進入spark_home]
bin/run-example streaming.DirectKafkaWordCount localhost:9092 sparkstreaming
結(jié)果能看到第六步對應的終端會出現(xiàn)一下情況,說明配置測試成功
消費者終端顯示的結(jié)果
spark-streaming運行結(jié)果
從結(jié)果來看涨享，網(wǎng)絡延遲大概有20秒左右。
附錄官網(wǎng)以kafka數(shù)據(jù)源的spark-streaming例子
streaming.DirectKafkaWordCount.scala
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// scalastyle:off println
package org.apache.spark.examples.streaming
import kafka.serializer.StringDecoder
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.SparkConf
/**
* Consumes messages from one or more topics in Kafka and does wordcount.
* Usage: DirectKafkaWordCount <brokers> <topics>
* <brokers> is a list of one or more Kafka brokers
* <topics> is a list of one or more kafka topics to consume from
*
* Example:
* $ bin/run-example streaming.DirectKafkaWordCount broker1-host:port,broker2-host:port \
* topic1,topic2
*/
object DirectKafkaWordCount {
def main(args: Array[String]) {
if (args.length < 2) {
System.err.println(s"""
|Usage: DirectKafkaWordCount <brokers> <topics>
| <brokers> is a list of one or more Kafka brokers
| <topics> is a list of one or more kafka topics to consume from
|
""".stripMargin)
System.exit(1)
}
StreamingExamples.setStreamingLogLevels()
val Array(brokers, topics) = args
// Create context with 2 second batch interval
val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount")
val ssc = new StreamingContext(sparkConf, Seconds(2))
// Create direct kafka stream with brokers and topics
val topicsSet = topics.split(",").toSet
val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
ssc, kafkaParams, topicsSet)
// Get the lines, split them into words, count the words and print
val lines = messages.map(_._2)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)
wordCounts.print()
// Start the computation
ssc.start()
ssc.awaitTermination()
}
}
