日志服务（SLS）集成 Spark 流计算实战

时间  2020-01-24
标签 日志 服务 sls 集成 spark 计算 实战 栏目 Spark 繁體版
原文   原文链接

前言

日志服务做为一站式的日志的采集与分析平台，提供了各类用户场景的日志采集能力，经过日志服务提供的各类与·与SDK，采集客户端（Logtail），Producer，用户能够很是容易的把各类数据源中的数据采集到日志服务的Logstore中。同时为了便于用户对日志进行处理，提供了各类支持流式消费的SDK，如各类语言的消费组，与 Spark，Flink，Storm 等各类流计算技术无缝对接的Connector，以便于用户根据本身的业务场景很是便捷的处理海量日志。html

从最先的Spark Streaming到最新的Stuctured Streaming，Spark 一直是最流行的流计算框架之一。使用日志服务的Spark SDK，能够很是方便的在Spark 中消费日志服务中的数据，同时也支持将 Spark 的计算结果写入日志服务。git

日志服务基础概念

日志服务的存储层是一个相似Kafka的Append only的FIFO消息队列，包含以下基本概念：github

  • 日志（Log）：由时间、及一组不定个数的Key-Value对组成。
  • 日志组（LogGroup）：一组日志的集合，包含相同Meta信息如Topic，Source，Tags等。是读写的基本单位。

  • Shard：分区，LogGroup读写基本单元，对应于Kafka的partition。
  • Logstore：日志库，用以存放同一类日志数据。Logstore会包含1个或多个Shard。
  • Project：Logstore存放容器，包含一个或者多个Logstore。

准备工做

1）添加Maven依赖：sql

<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。apache

对 Spark Streaming 的支持

Spark Streaming是Spark最先推出的流计算技术，如今已经进入维护状态，再也不会增长新的功能。可是考虑到Spark Streaming 的使用仍然很是普遍，咱们先从Spark Streaming开始介绍。Spark Streaming 提供了一个DStream 的数据模型抽象，本质是把无界数据集拆分红一个一个的RDD，转化为有界数据集的流式计算。每一个批次处理的数据就是这段时间内从日志服务消费到的数据。并发

图-2 DStream

Spark Streaming 从日志服务消费支持 Receiver 和 Direct 两种消费方式。app

Receiver模式

Receivers的实现内部实现基于日志服务的消费组（Consumer Library）。数据拉取与处理彻底分离。消费组自动均匀分配Logstore内的全部shard到全部的Receiver，而且自动提交checkpoint到SLS。这就意味着Logstore内的shard个数与Spark 实际的并发没有对应关系。
对于全部的Receiver，接收到的数据默认会保存在Spark Executors中，因此Failover的时候有可能形成数据丢失，这个时候就须要开启WAL日志，Failover的时候能够从WAL中恢复，防止丢失数据。框架

SDK将SLS中的每行日志解析为JSON字符串形式，Receiver使用示例以下所示：ide

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有以下的优点：

  1. 下降资源消耗，不须要占用Executor资源来做为Receiver的角色。
  2. 鲁棒性更好，在计算的时候才会从服务端真正消费数据，下降内存使用，再也不须要WAL，Failover 直接在读一次就好了，更容易实现exactly once语义。
  3. 简化并行。Spark partition 与 Logstore 的 shard 个数对应，增长shard个数就能提升Spark任务处理并发上限。

可是也存在一些缺点：

  1. 在SLS场景下，须要依赖本地的 ZooKeeper 来保存临时 checkpoint，当调用 commitAsync 时从 ZooKeeper同步到日志服务服务端。因此当须要重置 checkpoint 时，也须要先删除本地 ZooKeeper 中的 checkpoint 才能生效。
  2. 上一个批次保存 checkpoint 以前，下一个批次没法真正开始，不然 ZooKeeper 中的 checkpoint 可能会被更新成一个中间状态。目前SDK在每一个批次会检查是否上一个批次的 checkpoint 尚未提交，若是没有提交则生成一个空批次，而不是继续从服务端消费。
  3. 在 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主要有以下特色：

  1. 底层实现基于Spark SQL引擎，可使用大多数Spark SQL的函数。和Spark SQL共用大部分API，若是对Spark SQL熟悉的用户，很是容易上手。复用Spark SQL的执行引用，性能更佳。
  2. 支持 Process time 和 Event time，而Spark Streaming只支持 Process Time。
  3. 批流同一的API。Structured Streaming 复用Spark SQL的 DataSet/DataFrame模型，和 RDD/DStream相比更High level，易用性更好。
  4. 实时性更好，默认基于micro-batch模式。在 Spark 2.3 中，还增长了连续处理模型，号称能够作到毫秒级延迟。
  5. 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便可。

不足之处

  1. 不支持手动提交checkpoint，SDK内部自动保存checkpoint到checkpointLocation中。
  2. 再也不须要提供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

本文做者：liketic

阅读原文

本文为阿里云内容，未经容许不得转载。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程