公司业务准备上流数据处理了。由于之前基础平台选用了CDH，而CDH自带Spark，且由于数据源是每隔几分钟发一组数据文件的形式来传送数据，所以最终选取用Spark Steaming来做流数据处理。

下面记录初步使用Spark Steaming和Flume的一些过程。

第一个测试：Flume（spooldir to hdfs）

原始数据通过ftp每隔几分钟拉取一批数据到本地某文件夹。于是测试了下flume监控文件夹并将新加入的文件写入hdfs的功能。

配置文件如下：

根据官方文档hdfs.fileType默认是SequenceFile，这里选用DataStream将不压缩输出文件。

若不设置hdfs.useLocalTimeStamp为true则会报下面的错误，暂时不知为何。

期间还遇到了下面这个错误，原因很简单，当时想把路径设置成为yyyy-MM-dd hh:mm:ss这样的形式，但HDFS不允许创建带有“:”符号的文件夹，所以报错。

启动命令：

第二个测试：执行Spark Steaming任务

测试程序：

打jar包后放在集群上跑，结果第一次报错VERSION版本号不对，然后看了看本机编译的时候采用的是JAVA 8，而集群都是JAVA 7，于是重新用JAVA 7打包。第二次报错，大概意思是找不到org.apache.spark.streaming下面的方法，这个就无语了，尝试了多次后突然想到可能是所下载的spark lib包是java 8编译的，于是尝试了用CDH集群的spark lib包，好使，果然是lib包的问题。

出现各种找不到方法的错误，如下：

解决方法1：启动命令里面添加缺少的各种jar包

解决方法2：程序中添加所需的各种jar包

终于在启动时加上各种各样的jar包后可以正常的运行且不在报错，此时ctrl+c无法取消任务，且用mapred job -kill jobid也无法杀死任务，只能用yarn application -kill jobid来干掉任务。

第三个测试：Flume发送数据给Spark Streaming程序

Spark Streaming接收Flume以avro形式发送的数据。Flume配置文件如下：

先启动Spark Steaming任务，在启动Flume-ng，否则会报以下错误：

尝试将一个文件复制到Flume监控的/root/data文件夹下，Spark任务就会在log日志中输出一行：

但看了下放进去的文件行数是74，这里不知道为啥会多一行，不知道是不是文件头信息还是时间戳信息什么的，待后续分析。

参考资料

• Flume 1.6.0 User Guide
• Spark Overview
• Spark Streaming Programming Guide
• Spark Streaming + Flume Integration Guide

后续

开始看Spark Streaming的官方文档了，发现可以直接使用Spark Streaming中的Basic Sources的File Streams即可，这个玩意就能监控文件夹。看来用Flume有点多此一举了，具体如何取舍看实际业务需求。