日志服务(SLS)集成 Spark 流计算实战
前言
日志服务作为一站式的日志的采集与分析平台,提供了各种用户场景的日志采集能力,通过日志服务提供的各种与·与SDK,采集客户端(Logtail),Producer,用户可以非常容易的把各种数据源中的数据采集到日志服务的Logstore中。同时为了便于用户对日志进行处理,提供了各种支持流式消费的SDK,如各种语言的消费组,与 Spark,Flink,Storm 等各种流计算技术无缝对接的Connector,以便于用户根据自己的业务场景非常便捷的处理海量日志。
从最早的Spark Streaming到最新的Stuctured Streaming,Spark 一直是最流行的流计算框架之一。使用日志服务的Spark SDK,可以非常方便的在Spark 中消费日志服务中的数据,同时也支持将 Spark 的计算结果写入日志服务。
日志服务基础概念
日志服务的存储层是一个类似Kafka的Append only的FIFO消息队列,包含如下基本概念:
- 日志(Log):由时间、及一组不定个数的Key-Value对组成。
- 日志组(LogGroup):一组日志的集合,包含相同Meta信息如Topic,Source,Tags等。是读写的基本单位。
图-1 Log与LogGroup的关系
- Shard:分区,LogGroup读写基本单元,对应于Kafka的partition。
- Logstore:日志库,用以存放同一类日志数据。Logstore会包含1个或多个Shard。
- Project:Logstore存放容器,包含一个或者多个Logstore。
准备工作
1)添加Maven依赖:
<dependency> <groupId>com.aliyun.emr</groupId> <artifactId>emr-logservice_2.11</artifactId> <version>1.9.0</version> </dependency>
Github源码下载。
2)计划消费的日志服务project,logstore以及对应的endpoint。
3)用于访问日志服务Open API的Access Key。
对 Spark Streaming 的支持
Spark Streaming是Spark最早推出的流计算技术,现在已经进入维护状态,不再会增加新的功能。但是考虑到Spark Streaming 的使用仍然非常广泛,我们先从Spark Streaming开始介绍。Spark Streaming 提供了一个DStream 的数据模型抽象,本质是把无界数据集拆分成一个一个的RDD,转化为有界数据集的流式计算。每个批次处理的数据就是这段时间内从日志服务消费到的数据。
图-2 DStream
Spark Streaming 从日志服务消费支持 Receiver 和 Direct 两种消费方式。
Receiver模式
Receivers的实现内部实现基于日志服务的消费组(Consumer Library)。数据拉取与处理完全分离。消费组自动均匀分配Logstore内的所有shard到所有的Receiver,并且自动提交checkpoint到SLS。这就意味着Logstore内的shard个数与Spark 实际的并发没有对应关系。
对于所有的Receiver,接收到的数据默认会保存在Spark Executors中,所以Failover的时候有可能造成数据丢失,这个时候就需要开启WAL日志,Failover的时候可以从WAL中恢复,防止丢失数据。
SDK将SLS中的每行日志解析为JSON字符串形式,Receiver使用示例如下所示:
object SLSReceiverSample { def main(args: Array[String]): Unit = { val project = "your project" val logstore = "your logstore" val consumerGroup = "consumer group" val endpoint = "your endpoint" val accessKeyId = "access key id" val accessKeySecret = "access key secret" val batchInterval = Milliseconds(5 * 1000) val conf = new SparkConf().setAppName("Test SLS Loghub") val ssc = new StreamingContext(conf, batchInterval) val stream = LoghubUtils.createStream( ssc, project, logstore, consumerGroup, endpoint, accessKeyId, accessKeySecret, StorageLevel.MEMORY_AND_DISK, LogHubCursorPosition.END_CURSOR) stream.checkpoint(batchInterval * 2).foreachRDD(rdd => rdd.map(bytes => new String(bytes)).top(10).foreach(println) ) ssc.checkpoint("hdfs:///tmp/spark/streaming") ssc.start() ssc.awaitTermination() } }
除Project,Logstore,Access Key 这些基础配置外,还可以指定StorageLevel,消费开始位置等。
Direct模式
Direct模式不再需要Receiver,也不依赖于消费组,而是使用日志服务的低级API,在每个批次内直接从服务端拉取数据处理。对于Logstore中的每个Shard来说,每个批次都会读取指定位置范围内的数据。为了保证一致性,只有在每个批次确认正常结束之后才能把每个Shard的消费结束位置(checkpoint)保存到服务端。
为了实现Direct模式,SDK依赖一个本地的ZooKeeper,每个shard的checkpoint会临时保存到本地的ZooKeeper,等用户手动提交checkpoint时,再从ZooKeeper中同步到服务端。Failover时也是先从本地ZooKeeper中尝试读上一次的checkpoint,如果没有读到再从服务端获取。
object SLSDirectSample { def main(args: Array[String]): Unit = { val project = "your project" val logstore = "your logstore" val consumerGroup = "consumerGroup" val endpoint = "endpoint" val accessKeyId = "access key id" val accessKeySecret = "access key secret" val batchInterval = Milliseconds(5 * 1000) val zkAddress = "localhost:2181" val conf = new SparkConf().setAppName("Test Direct SLS Loghub") val ssc = new StreamingContext(conf, batchInterval) val zkParas = Map("zookeeper.connect" -> zkAddress) val loghubStream = LoghubUtils.createDirectStream( ssc, project, logstore, consumerGroup, accessKeyId, accessKeySecret, endpoint, zkParas, LogHubCursorPosition.END_CURSOR) loghubStream.checkpoint(batchInterval).foreachRDD(rdd => { println(s"count by key: ${rdd.map(s => { s.sorted (s.length, s) }).countByKey().size}") // 手动更新checkpoint loghubStream.asInstanceOf[CanCommitOffsets].commitAsync() }) ssc.checkpoint("hdfs:///tmp/spark/streaming") // set checkpoint directory ssc.start() ssc.awaitTermination() } }
Direct模式示例
如何限速
在Receiver中,如果需要限制消费速度,我们只需要调整 Consumer Library 本身的参数即可。而Direct方式是在每个批次开始时从SLS拉取数据,这就涉及到一个问题:一个批次内拉取多少数据才合适。如果太多,一个批次内处理不完,造成处理延时。如果太少会导worker空闲,工作不饱和,消费延时。这个时候我们就需要合理配置拉取的速度和行数,实现一个批次尽可能多处理又能及时完成的目标。理想状态下Spark 消费的整体速率应该与SLS采集速率一致,才能实现真正的实时处理。
由于SLS的数据模型是以LogGroup作为读写的基本单位,而一个LogGroup中可能包含上万行日志,这就意味着Spark中直接限制每个批次的行数难以实现。因此,Direct限流涉及到两个配置参数:
参数 | 说明 | 默认值 |
---|---|---|
spark.streaming.loghub.maxRatePerShard | 每个批次每个Shard读取行数,决定了限流的下限 | 10000 |
spark.loghub.batchGet.step | 每次请求读取LogGroup个数,决定了限流的粒度 | 100 |
可以通过适当缩小spark.loghub.batchGet.step来控制限流的精度,但是即便如此,在某些情况下还是会存在较大误差,如一个LogGroup中存在10000行日志,spark.streaming.loghub.maxRatePerShard设置为100,spark.loghub.batchGet.step设置为1,那一个批次内该shard还是会拉取10000行日志。
两种模式的对比
和Receiver相比,Direct有如下的优势:
- 降低资源消耗,不需要占用Executor资源来作为Receiver的角色。
- 鲁棒性更好,在计算的时候才会从服务端真正消费数据,降低内存使用,不再需要WAL,Failover 直接在读一次就行了,更容易实现exactly once语义。
- 简化并行。Spark partition 与 Logstore 的 shard 个数对应,增加shard个数就能提高Spark任务处理并发上限。
但是也存在一些缺点:
- 在SLS场景下,需要依赖本地的 ZooKeeper 来保存临时 checkpoint,当调用 commitAsync 时从 ZooKeeper同步到日志服务服务端。所以当需要重置 checkpoint 时,也需要先删除本地 ZooKeeper 中的 checkpoint 才能生效。
- 上一个批次保存 checkpoint 之前,下一个批次无法真正开始,否则 ZooKeeper 中的 checkpoint 可能会被更新成一个中间状态。目前SDK在每个批次会检查是否上一个批次的 checkpoint 还没有提交,如果没有提交则生成一个空批次,而不是继续从服务端消费。
- 在 SLS 场景下,限流方式不够精确。
Spark Streaming结果写入SLS
与消费SLS相反,Spark Streaming的处理结果也可以直接写入SLS。使用示例:
... val lines = loghubStream.map(x => x) // 转换函数把结果中每条记录转为一行日志 def transformFunc(x: String): LogItem = { val r = new LogItem() r.PushBack("key", x) r } val callback = new Callback with Serializable { override def onCompletion(result: Result): Unit = { println(s"Send result ${result.isSuccessful}") } } // SLS producer config val producerConfig = Map( "sls.project" -> loghubProject, "sls.logstore" -> targetLogstore, "access.key.id" -> accessKeyId, "access.key.secret" -> accessKeySecret, "sls.endpoint" -> endpoint, "sls.ioThreadCount" -> "2" ) lines.writeToLoghub( producerConfig, "topic", "streaming", transformFunc, Option.apply(callback)) ssc.checkpoint("hdfs:///tmp/spark/streaming") // set checkpoint directory ssc.start() ssc.awaitTermination()
对Structured Streaming的支持
Structured Streaming 并不是最近才出现的技术,而是早在16年就已经出现,但是直到 Spark 2.2.0 才正式推出。其数据模型是基于无界表的概念,流数据相当于往一个表上不断追加行。
图-3 无界表模型
与Spark Streaming相比,Structured Streaming主要有如下特点:
- 底层实现基于Spark SQL引擎,可以使用大多数Spark SQL的函数。和Spark SQL共用大部分API,如果对Spark SQL熟悉的用户,非常容易上手。复用Spark SQL的执行引用,性能更佳。
- 支持 Process time 和 Event time,而Spark Streaming只支持 Process Time。
- 批流同一的API。Structured Streaming 复用Spark SQL的 DataSet/DataFrame模型,和 RDD/DStream相比更High level,易用性更好。
- 实时性更好,默认基于micro-batch模式。在 Spark 2.3 中,还增加了连续处理模型,号称可以做到毫秒级延迟。
- API 对用户更友好,只保留了SparkSession一个入口,不需要创建各种Context对象,使用起来更简单。
SDK使用示例
import org.apache.spark.sql.SparkSession import org.apache.spark.sql.types.{StringType, StructField, StructType} object StructuredStreamingDemo { def main(args: Array[String]) { val spark = SparkSession .builder .appName("StructuredLoghubWordCount") .master("local") .getOrCreate() import spark.implicits._ val schema = new StructType( Array(StructField("content", StringType))) val lines = spark .readStream .format("loghub") .schema(schema) .option("sls.project", "your project") .option("sls.store", "your logstore") .option("access.key.id", "your access key id") .option("access.key.secret", "your access key secret") .option("endpoint", "your endpoint") .option("startingoffsets", "latest") .load() .select("content") .as[String] val wordCounts = lines.flatMap(_.split(" ")).groupBy("value").count() val query = wordCounts.writeStream .outputMode("complete") .format("loghub") .option("sls.project", "sink project") .option("sls.store", "sink logstore") .option("access.key.id", "your access key id") .option("access.key.secret", "your access key secret") .option("endpoint", "your endpoint") .option("checkpointLocation", "your checkpoint dir") .start() query.awaitTermination() } }
代码解释:
1)schema 声明了我们需要的字段,除了日志中的字段外,还有如下的内部字段:
__logProject__ __logStore__ __shard__ __time__ __topic__ __source__ __sequence_number__ // 每行日志唯一id
如果没有指定schema,SDK默认提供一个__value__字段,其内容为由所有字段组成的一个JSON字符串。
2)lines 定义了一个流。
startingoffsets:开始位置,支持:
- latest :日志服务最新写入位置。强烈建议从latest开始,从其他位置开始意味着需要先处理历史数据,可能需要等待较长时间才能结束。
- earliest:日志服务中最早的日志对应的位置。
- 或者为每个shard指定一个开始时间,以JSON形式指定。
maxOffsetsPerTrigger:批次读取行数,SDK中默认是64*1024 。
3)结果写入到日志服务
format 指定为Loghub即可。
不足之处
- 不支持手动提交checkpoint,SDK内部自动保存checkpoint到checkpointLocation中。
- 不再需要提供consumerGroup名称,也就是说checkpoint没有保存到SLS服务端,无法在日志服务里面监控消费延迟,只能通过Spark 任务日志观察消费进度。
参考资料
官方文档:https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html
SLS SDK例子:https://github.com/aliyun/aliyun-emapreduce-sdk/tree/master-2.x/examples/src/main/scala/com/aliyun/emr/examples/sql/streaming
日志服务实时消费:https://help.aliyun.com/document_detail/28998.html
欢迎扫群加入日志服务技术交流钉钉群
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
01月01日云栖号头条:中国高铁进入“智能自动驾驶”时代
云栖号:https://www.aliyun.com/#module-yedOfott8第一手的上云资讯,不同行业精选的上云企业案例库,基于众多成功案例萃取而成的最佳实践,助力您上云决策! 今日最新云头条快讯:这项具有历史意义的立法将为美国消费者提供更强大的保护,从而使他们免受不请自来的骚扰电话的骚扰。,一起来看最新的资讯: 国家智能网联汽车创新中心在京启动建设 国家智能网联汽车创新中心建设启动会暨车联网(智能网联汽车)和自动驾驶地图应用试点签约活动在北京召开。工信部副部长王志军要求,着力解决智能网联汽车发展的关键共性技术问题。 反机器人电话法案《TRACED》正式成为法律 据外媒报道,美国总统唐纳德·特朗普日前签署了反机器人电话法《Pallone-Thune TRACED(全称Pallone-Thune Telephone Robocall Abuse Criminal Enforcement and Deterrence)法案》使之成为法律,进而扩大了FCC在执法和可能的罚款方面的权力。该法在参议院和众议院中几乎得到了两党一致的支持。声明还称:“美国家庭应该拥有控制他们自己通讯录...
- 下一篇
将ERP与物联网技术集成的5大好处
云栖号:https://www.aliyun.com/#module-yedOfott8第一手的上云资讯,不同行业精选的上云企业案例库,基于众多成功案例萃取而成的最佳实践,助力您上云决策! 图片来源:pixabay 来源:物联之家网(iothome.com) 现代化的ERP系统将所有业务应用无缝连接到一个统一的平台中,从而提高了数据的可见性、控制和决策能力。然而,如今的ERP业务流程与几十年前的设计方式相比基本没有什么变化。但是,将ERP与物联网(IoT)集成会非常有益——对于后台以及面向客户的流程都是如此。 物联网解决方案允许将各种外部机器和设备的数据实时与ERP系统共享,使公司无需费力地将其内部业务应用和流程与外部数据连接起来。从这些物联网设备收集的数据是实时的,而且非常准确,从而增加了获得可操作见解的机会。 将ERP与物联网集成的好处是什么? 由于提高了数据质量和可用性,因此可以更好地进行决策 Gartner估计,到2020年,将有超过260亿台物联网设备投入使用。这意味着公司可用的数据量正在迅速增长,但许多收集这些数据的公司还没有准备好应对大量数据所带来的挑战。成功的数字转型需...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- CentOS6,CentOS7官方镜像安装Oracle11G
- Red5直播服务器,属于Java语言的直播服务器
- Hadoop3单机部署,实现最简伪集群
- CentOS7,CentOS8安装Elasticsearch6.8.6
- CentOS7安装Docker,走上虚拟化容器引擎之路
- CentOS7编译安装Gcc9.2.0,解决mysql等软件编译问题
- Windows10,CentOS7,CentOS8安装MongoDB4.0.16
- CentOS7编译安装Cmake3.16.3,解决mysql等软件编译问题
- SpringBoot2整合Redis,开启缓存,提高访问速度
- CentOS6,7,8上安装Nginx,支持https2.0的开启