spark streaming + flume+python(编程)初探

一、环境部署

hadoop集群2.7.1
flume 1.7.0
spark集群:spark-2.0.1-bin-hadoop2.7.tgz
环境搭建可参考我前面几篇文章。不再赘述
三台机器:master,slave1,slave2

二、启动集群环境

1.启动hadoop集群
start-all.sh
2.启动spark集群
start-master.sh
start-slaves.sh

三、配置flume

编辑conf/flume-conf.properties
配置source、channel、sink
sink是关键
一种是推式接收器
一种是拉式接收器
相对于的spark streaming 读取代码也要相应改变
这里我采用的是推式接收器
sink则应这样配置

a1.sinks = k1
a1.sinks.k1.type = avro  
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = master  #绑定的主机名
a1.sinks.k1.port = 9999  #绑定的端口号```
source可谓avro,syslog等等,第一次测试为确保成功,可先考虑avro,与sink对应
**avro:**

a1.sources.r1.type = avro
a1.sources.r1.bind = 192.168.31.131
a1.sources.r1.port = 4141
a1.sources.r1.channels = c1

**syslog:**

a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 5140
a1.sources.r1.host = 192.168.31.131
a1.sources.r1.channels = c1

channel均为内存

最后完整配置如下:
source为avro:

a1.sources = r1
a1.sinks = k1
a1.channels = c1

Describe/configure the source

a1.sources.r1.type = avro
a1.sources.r1.bind = 192.168.31.131
a1.sources.r1.port = 4141
a1.sources.r1.channels = c1

Describe the sink

a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = master
a1.sinks.k1.port = 9999

Use a channel which buffers events in memory

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

yi bai tiao jiu submit

source 为syslog:

a1.sources = r1
a1.sinks = k1
a1.channels = c1

Describe/configure the source

a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 5140
a1.sources.r1.host = 192.168.31.131
a1.sources.r1.channels = c1

Describe the sink

a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = master
a1.sinks.k1.port = 9999

Use a channel which buffers events in memory

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

yi bai tiao jiu submit

####四、编程FlumeWordCount.py
**编写spark steaming 代码,读取flume采集的数据,并统计词频**

代码(python 实现):
```
# -*- coding: UTF-8 -*-
###spark streaming&&Flume 
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.flume import FlumeUtils

sc=SparkContext("local[2]","FlumeWordCount")
#处理时间间隔为2s
ssc=StreamingContext(sc,2)

#打开一个TCP socket 地址 和 端口号
#lines=ssc.socketTextStream("master",9999)
lines = FlumeUtils.createStream(ssc, "master",9999)
lines1=lines.map(lambda x:x[1])
#对1s内收到的字符串进行分割
words=lines1.flatMap(lambda line:line.split(" "))

#映射为(word,1)元祖
pairs=words.map(lambda word:(word,1))

wordcounts=pairs.reduceByKey(lambda x,y:x+y)

#输出文件,前缀+自动加日期
wordcounts.saveAsTextFiles("/tmp/flume")

wordcounts.pprint()

#启动spark streaming应用
ssc.start()
#等待计算终止
ssc.awaitTermination()
```
####五、运行
#####1.下载依赖的jars包
注意,应该去官网找对应的jar包,例如
kafka2.01对应

![](http://upload-images.jianshu.io/upload_images/1908836-042927ced5e8bf74.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
下载spark-streaming-flume_2.11.jar 我放在了/home/cms下
(ps:之前放在了flume/lib/下,就会报找不到classpath的错
#####2.启动flume agent 
```
flume-ng agent --conf ./conf/ -f /home/cms/flume/conf/flume-conf.properties -n a1 -Dflume.root.logger=INFO,console
```
>--conf对于的是flume下的conf目录-f 对应的是agent配置文件-n 对于的是agent的名字
基本上终端都会报错 ,因为没有启动client和sink落地的程序,暂时不用理

#####3.**若source为avro:**,启动测试
1)准备测试数据
新建一个log_test.txt
输入数据:

![](http://upload-images.jianshu.io/upload_images/1908836-57970eaa4c5f7e80.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

2)运行spark streaming
开启另一个终端
```
 spark-submit --jars spark-streaming-flume-assembly_2.11-2.0.1.jar FlumeWordCount.py 2> error_log.txt```

3)发送数据
开启另外一个终端,将log_test.txt发送出去
```
flume-ng  avro-client --conf ./conf/ -H 192.168.31.131 -p 4141 -F /home/cms/log_test.txt```

![](http://upload-images.jianshu.io/upload_images/1908836-4e64718c4b7a6436.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

4)观察运行spark streaming的程序的终端的输出

![](http://upload-images.jianshu.io/upload_images/1908836-ce0b33bc0a01e0d7.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
若只是输出time。。。等等,则检查flume的配置
hdfs上查看:

![](http://upload-images.jianshu.io/upload_images/1908836-4f413506d9c19af5.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

![](http://upload-images.jianshu.io/upload_images/1908836-40e7a2bd0e9b3516.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

![](http://upload-images.jianshu.io/upload_images/1908836-d8b36c1143406ad3.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

![](http://upload-images.jianshu.io/upload_images/1908836-7e347d968cb5302a.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
#####4.**若source为syslog:**,启动测试
1)运行spark streaming 程序
```
 spark-submit --jars spark-streaming-flume-assembly_2.11-2.0.1.jar FlumeWordCount.py 2> error_log.txt```
2)开启另一个终端,发送数据
```
echo "hello'\t'word" | nc 192.168.31.131 5140
```

![](http://upload-images.jianshu.io/upload_images/1908836-44c4113f614dae33.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
3)观察运行spark streaming的程序的终端的输出

![](http://upload-images.jianshu.io/upload_images/1908836-1ff470da2eb4603f.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

####四、下一步
flume+kafka+spark streaming

参考:[官网](http://spark.apache.org/docs/latest/streaming-flume-integration.html)
    原文作者:玄月府的小妖在debug
    原文地址: https://www.jianshu.com/p/23a906d5e59f
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞