之前我在微信朋友圈发了一段话,说明Spark Streaming 不仅仅是流式计算,也是一类通用的模式,可以让你只关注业务逻辑而无需关注分布式相关的问题而迅速解决业务问题
前两天我刚在自己的一篇文章中鼓吹数据天生就是流式的,并且指出:
批量计算已经在慢慢退化,未来必然是属于流式计算的,数据的流动必定是由数据自己驱动流转的。
而Spark Streaming 在上层概念上,完美融合了批量计算和流式计算,让他们你中有我,我中有你,这种设计使得Spark Streaming 作为流式计算的一个载体,同时也能作为其他一些需要分布式架构的问题提供解决方案。
现在以标题中的采集系统为例,整个事情你只要实现采集逻辑,至于具体元数据读取,结果存储到哪都可能只要个简单配置或者利用现成的组件,最后部署也只要简单申明下资源就可以在一个可以弹性扩展的集群上。
关于这块的理念,可参考
目前这个采集系统主要是为了监控使用。但凡一个公司,或者部门内部会有大量的开源系统,每个开源组件都会提供大致三类输出:
但是对于监控来说,前面两个直观易用,但是也都有比较大的问题:
相反,Rest 接口最为灵活,但是需要自己做写逻辑,比如获取数据,处理,然后做自己的呈现 。问题来了,如果我现在有几千个Rest接口的数据要获取,并且需要一个很方便的手段抽取里面要的值(或者指标)。这便涉及到了两个问题:
QQ20160529-0@2x.png
结构
回到上面的一个问题,
接口返回的数据形态各异,如何提供一个方便一致的模型,让用户简单通过一个配置就可以抽取出里面的内容
Rest 接口返回的数据,无非四种:
对于1,我们先不探讨。对于JSON,XML 我们可以采用 XPATH,对于TEXT我们可以采用标准的正则或者ETL来进行抽取。
我们在定义一个需要采集的URL时,需要同时配置需要采集的指标以及对应的指标的XPATH路径或者正则。当然也可以交给后端的ETL完成该逻辑。不过我们既然已经基于Spark Streaming做采集系统,自然也可以利用其强大的数据处理功能完成必要的格式化动作。所以我们建议在采集系统直接完成。
数据源的一个可能的数据结构:
appName 采集的应用名称,cluster1,cluster2 appType 采集的应用类型,storm/zookeeper/yarn 等 url 需要采集的接口 params 可能存在例如post请求之类的,需要额外的请求参数 method Get/POST/PUT 等请求方法体 key_search_qps : $.store.book[0].author 定义需要抽取的指标名称以及在Response 对应的XPATH 路径 key_..... 可以有更多的XPATH key_..... 可以有更多的XPATH extraParams 人工填写一些其他参数
采集系统通过我们封装的一个 DInputStream,然后根据batch(调度周期),获取这些数据,之后交给特定的执行逻辑去执行。采用 StreamingPro ,会是这样:
"RestCatch": { "desc": "RestCatch", "strategy": "....SparkStreamingStrategy", "algorithm": [], "ref": [], "compositor": [ { "name": "....ESInputCompositor", "params": [ { "es.nodes": "....", "es.resource": "monitor_rest/rest" } ] }, { "name": ".....RestFetchCompositor",//发起http请求,获取response "params": [ { "resultKey": "result", "keyPrefix": "key_" } ] }, { "name": "....JSonExtractCompositor",//根据XPath获取response里的指标 "params": [ { "resultKey": "result", "keyPrefix": "key_" } ] }, { "name": ".....ConsoleOutputCompositor",//输出结果 "params": [] } ], "configParams": { } }
通过上面的配置文件,可以很好看到处理流程。
元数据管理系统是必要的,他可以方便你添加新的URL监控项。通过 StreamingPro ,你可以在Spark Streaming 的Driver中添加元数据管理页面,实现对元数据的操作逻辑。我们未来会为 如何通过StreamingPro
给Spark Streaming 添加自定义Rest 接口/Web页面提供更好的教程。
上面其实已经是试下了一个采集系统的雏形,得益于Spark Streaming天然的分布式,以及灵活的算子,我们的系统是足够灵活,并且可横向扩展。
然而你会发现,
第一个问题很好解决,我们在元数据里定义采集周期,而Spark Streaming的调度周期则设置为最小粒度。
第二个问题容错性属于业务层面的东西,但是如果有Task失败,Spark Streaming也会把你尝试重新调度和重试。我们建议由自己来完成。
第三个,只要开启了 Dynamic Resource Allocation,则能够根据情况,实现资源的伸缩利用。