开发者学堂课程【快速掌握阿里云 E-MapReduce： E-Mapreduce 流式处理】学习笔记与课程紧密联系，让用户快速学习知识
课程地址：https://developer.aliyun.com/learning/course/390/detail/5006
E-Mapreduce 流式处理
目录：
一、 回顾内容
二、 基本流程
三、 基本架构
四、 环境准备
五、 下载及编译
六、 配置 SLS 源
七、 Local 执行-调试程序
一、内容回顾
Spark Streaming 是 EM 提供的 spark 上的框架，是处理流式计算的。Log Service是一个日志采集的服务。
二、基本流程
流程如下：
1. 创建集群。
2. 下载、编译代码。
3. 本地执行。（本地可以 log 跑 Spark Streaming）
4. 上传 jar、资源文件。
5. 创建作业。
6. 新建执行计划。
7. 执行。
8. 查看结果/看日志。
9. 通过网页看下作业的运行情况。
三、基本架构
基本架构是 EM 里面的 Spark Streaming 会去消费 Log Service 的采集日志。如下图所示。
四、 环境准备
WIN 环境准备:
1.下载 git，下载地址:https://git-scm.com/download/win
2.准备 IDE，以 IDEA 为例子，下载地址:
https://www.jetbrains.com/idea/download/#section=windows
3.JRE准备1.7版本(后续会支持1.8版本),下载地址:
http:/www.oracle.com/technetwork/cn/java/javase/downloads/jre7-downloads-1880261.html
4.maven，下载地址:https://maven.apache.org/download.cgi
五、下载及编译
首先下载下面的代码，然后编译代码：mvn install，将其可克隆之后编译即可。
下载代码:
git clone https://github.com/aliyun/aliyun-emapreduce-demo.git·
编译代码:
mvn install
六、配置SLS源
到 LogService 管控平台，创建一个 project,名称取为 teststreaming4，所属区域填为华东1（杭州），点确定。然后接着创建一个 LogStore，名称取为teststreaming4logstore。然后选择数据源，配置名称填为 fliter_info_count。再绑定手机模式,登录外网集群（114.55.40.172），然后登录 mast（mast 生成链路是为了产生载机）再登录 $ ssh root@114.55.40.172 进去 ，敲入代码
[rootbear-header-! 718 sudo su hadoop[hadoopbear header-1 root]s ed[hadooptear-headez-131s
[hadoopdear-header-1 78 cd logs/[hadoopeear-hesder-1 logs]s 11 total 56344
-IV-IV-I--1 hadoop hadoop 2645178 lay 15:47 top2.log
-IV-IV-I--1 hadoop hadoop 7540 lay 16:11 top3.log
-IV-IV-I--1 hadoop hedoop 55034042 lay 16:06 top.log
[hadooptear header-1 logs]$ top -d 1 >> top4.log
[hadoogtear header-1 logs]$ 1s
top2.log top3.log top4.log top.log
[hadoogtear-header-1 logs]$ cat C
[hadoopGcar header-1 logs]$ vi top4.log
[1]+ Stopped vin top4.log
[hadoop@ear-header-1 logs]$
[hadoop@ear-header-1 logs]$ 1s
top2.log top3.log top4.log top.log
[hadoogbear-header-1 logs]s 1l total 56352
-IV-IV-I--1 hadoop hedoop 2645178 lay 15:47 top2.log
-IV-IV-I--1 hadoop 7540 lay 16:11 top3.log
-IV-IV-I--1 hadoop hadoop 7532 lay 16:12 top4.log
-IV-IV-I--1 hadoop hadoop 55034042 lay 16:06 top.log
[hadoopOenr-hesder-1 log=]$ pvd hone/hadoop/logs
[hadooghear-header ogs]$）
得到目录后再将目录配到日志路径里，选择极简模式接着下一步。最后选择机器模式创建机器组，名称取为 emr4，填内网 IP 地址（10.25.0.137）。语言就创建好了。来到云监控控制台，如果里面没有数据，就换代码给一组数据。
七、 Local执行-调试程序
代码用 conf 加 must，会得到一个 loghubStream
1.参数:
LoghubSample:
teststreaming4 teststreaming4logstore fliter info count
testemrstream4.cn-hangzhou.log.aliyuncs.com 93recQArCQwaJPOd KHRaOzQgrJVZweuAVfEB4KNHDUlu6g 1
界面日志：
登录到 IDE，里面有 loghubsample.scala 工程
val conf =new SparkConf().setAppName("Loghub Sample") conf.setMaster("local).setAppName("test")
val ssc = new StreamingContext(conf, batchinterval)
val loghubStream =LoghubUtils. createStream( ssc,
loghubProject,
logStore,
loghubGroupName,
endpoint,
1,
accessKeyId,
accessKeySecret,
StorageLevel.MEMORY_AND_DISA)
loghubStream. foreachRDD(rdd => println(rdd. count ()))
ssc. Start()
ssc.awaitTermination()
用 conf 生成 setMaster,再得到 loghubStream,跑之前先改个配置。
数据在不停的跑着。
2.
在线上跑 Log，先配作业，集成根据自身情况可以适当减少，里面参数根据需求可以改一下。
然后创建集群计划，以 test 命名。刷新之后点执行，数据已经开始跑路。在 OSS管理控制台上看包，点击 object 管理，找到 shad 将其删除。因为包会依赖很多，所以直接用 shad。
然后将作业停止跑路。
3.创建执行计划：
(1) 选择集群方式
(2) 选择集群
(3) 配置调度的方法
(4) 配置作业
创建成功后，刷新执行程序就跑起来了，然后到 oss 管理控制台，来到 object 管理，删掉 shaded。先停掉作业，再刷新执行，最后打通 OSS 隧道。
$ ss-1 1d.rss -ND 8157 hadoope114.55.40.172
channel 19: open failed: edainistretively prohsbited: opep fssled
channel 20: open failed: adainistratively prohsbited: open fasled
channel 21: open failed: edainastratively! psohibifed: open fasled
channel 1: open failed: connect farled: Cormection tined out
channel 2: open failed: connect failed; Coraction tined out
channel 4: open failed: coenect failed: Connoction tined out channel10: open failsd: connect failed: Coenection tined out channel11: open failed: connect failed: Cormection tined out
channel 12: open failed: connect failed: Cormection tinod out
charnel13: open failed: connect failed: Cotnection tinod out charnel15:opea failed: connect failed: Coenection tined out channel5:open failed: ccnnect fasled: Corsecticn taned out channel6:cpen failed: cotnect fasled: Connecticn taned out
channel 7: open failed: ccenect fasled: Cennecticn tined out
channel 16:open failed: connect failed: Connection tined out channel 17:open failed: connect lasled: Coenection tined out
channel 4:open failed: ccenect failed: Conaectiod tined out
channe 9:open failed: ccanect farled: Conaection taned out
channel 3: open failed: ccanect fazled: Cormection tined out
channel 8: open failed: connect failed: Coraection taned out
channel 27: open failed: connect fsiled; Connecticn timed out channel 7: open failed: cornect failed: Coraection tined out channel 1: open failed: connect faled; Connection taned out
channel 2: open failed: cormect failed: Connection tined out
channel 5: open failed: coanect failed: Ccnnectaon taned out channel 10: open failed: connect failed: Counection tined out channel 4: open failed: connect failed; Cormection tined out channel 1: open failed: connect farled: Cornection tined out channel 9: open failed: connect falled: Conmection timed out channel 12: open failed: connect failed: Cornection tined out channel i: open failed: cornect farled: Connection tined out channel 15: open failed: connect fuuled: Connection tined out channel 1s open failed: connect farled: Connection tamed out
启动浏览器到Spark界面，Executars 和 Streaming 都在跑路。因为 Streaming 是长周期运行的，所以它在本地、线上都可以执行。由于过期问题，如果在点击时出现问题，刷新一下即可。
jobs：
Executors:
Streaming:
