val path = new File(".").getCanonicalPath()
//File workaround = new File(".");
System.getProperties().put("hadoop.home.dir", path);
new File("./bin").mkdirs();
new File("./bin/winutils.exe").createNewFile();
//val sparkConf = new SparkConf().setAppName("HdfsWordCount").setMaster("local[2]")
val sparkConf = new SparkConf().setAppName("HdfsWordCount")
// Create the context
val ssc = new StreamingContext(sparkConf, Seconds(20))
//val hostname = "127.0.0.1"
val hostname = "localhost"
val port = 2345
val storageLevel = StorageLevel.MEMORY_ONLY
val flumeStream = FlumeUtils.createStream(ssc, hostname, port, storageLevel)
flumeStream.count().map(cnt => "Received " + cnt + " flume events." ).print()
ssc.start()
ssc.awaitTermination()
}
}
flume配置文件如下:
# Please paste flume.conf here. Example:
# Sources, channels, and sinks are defined per
# agent name, in this case 'tier1'.
tier1.sources = source1
tier1.channels = channel1
tier1.sinks = sink1
# For each source, channel, and sink, set
# standard properties.
tier1.sources.source1.type = exec
tier1.sources.source1.command = tail -F /opt/data/test3/123
tier1.sources.source1.channels = channel1
tier1.channels.channel1.type = memory
#tier1.sinks.sink1.type = logger
tier1.sinks.sink1.type = avro
tier1.sinks.sink1.hostname = localhost
tier1.sinks.sink1.port = 2345
tier1.sinks.sink1.channel = channel1
# Other properties are specific to each type of yhx.hadoop.dn01
# source, channel, or sink. In this case, we
# specify the capacity of the memory channel.
tier1.channels.channel1.capacity = 100