Simple Log Service:Use o Spark Streaming para consumir dados de log

Limites

O consumo com o Spark Streaming é compatível apenas com o Spark 2.x.

O Alibaba Cloud Spark SDK oferece suporte a dois modos de consumo: Receiver e Direct. Adicione a seguinte dependência do Maven ao seu projeto:

<dependency>
  <groupId>com.aliyun.emr</groupId>
  <artifactId>emr-logservice_2.11</artifactId>
  <version>1.7.2</version>
</dependency>

Consuma dados de log no modo Receiver

Consuma dados de log no modo Direct

O modo Direct requer um service ZooKeeper para armazenar o estado intermediário, portanto defina um diretório de checkpoint no ZooKeeper para persistir os dados intermediários. Para reconsumir os dados após reiniciar uma tarefa, exclua o diretório correspondente no ZooKeeper e altere o nome do grupo de consumidores.

• Parâmetros

  Parâmetro	Tipo	Descrição
  project	String	O nome do projeto no SLS.
  logstore	String	O nome do logstore no SLS.
  consumerGroup	String	O nome do grupo de consumidores. Este nome serve apenas para salve os checkpoints de consumo.
  endpoint	String	O endpoint da região que hospeda o projeto do SLS. Para mais informações, consulte Endpoints.
  accessKeyId	String	O AccessKey ID para acesso ao SLS.
  accessKeySecret	String	O AccessKey Secret para acesso ao SLS.
  zkAddress	String	A URL de conexão do service ZooKeeper.

• Limitação de taxa

  O Spark Streaming processa dados em microlotes. Especifique o número de entradas de log consumidas por lote e por shard.

  No SLS, o sistema armazena cada solicitação de gravação como um grupo de logs. Uma solicitação de gravação típica contém várias entradas de log em um único grupo de logs. Ao usar o rastreamento web (web tracking), cada solicitação de gravação contém apenas uma entrada de log por grupo de logs. Os parâmetros a seguir controlam a quantidade de dados consumidos em um único lote.

  Parâmetro	Descrição	Padrão
  spark.loghub.batchGet.step	O número máximo de grupos de logs buscados por solicitação de consumo.	100
  spark.streaming.loghub.maxRatePerShard	O número máximo de entradas de log consumidas por shard e por lote.	10000

  O parâmetro spark.streaming.loghub.maxRatePerShard define o limite máximo de entradas de log por shard e por lote. O SDK busca grupos de logs em incrementos de spark.loghub.batchGet.step e acumula a contagem de entradas de log. Quando a contagem atinge ou excede o valor de spark.streaming.loghub.maxRatePerShard, o SDK interrompe a busca. Como a granularidade do consumo está no nível do grupo de logs, o limite de spark.streaming.loghub.maxRatePerShard é aproximado. A contagem real por lote depende de spark.loghub.batchGet.step e do número de entradas de log em cada grupo de logs.

• Exemplo

  import com.aliyun.openservices.loghub.client.config.LogHubCursorPosition
  import org.apache.spark.SparkConf
  import org.apache.spark.streaming.{ Milliseconds, StreamingContext}
  import org.apache.spark.streaming.aliyun.logservice.{ CanCommitOffsets, LoghubUtils}
  
  object TestDirectLoghub {
    def main(args: Array[String]): Unit = {
      if (args.length < 7) {
        System.err.println(
          """Usage: TestDirectLoghub <project> <logstore> <loghub group name> <endpoint>
            |         <access key id> <access key secret> <batch interval seconds> <zookeeper host:port=localhost:2181>
          """.stripMargin)
        System.exit(1)
      }
  
      val project = args(0)
      val logstore = args(1)
      val consumerGroup = args(2)
      val endpoint = args(3)
      val accessKeyId = args(4)
      val accessKeySecret = args(5)
      val batchInterval = Milliseconds(args(6).toInt * 1000)
      val zkAddress = if (args.length >= 8) args(7) else "localhost:2181"
  
      def functionToCreateContext(): StreamingContext = {
        val conf = new SparkConf().setAppName("Test Direct Loghub")
        val ssc = new StreamingContext(conf, batchInterval)
        val zkParas = Map("zookeeper.connect" -> zkAddress,
          "enable.auto.commit" -> "false")
        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}")
          loghubStream.asInstanceOf[CanCommitOffsets].commitAsync()
        })
        ssc.checkpoint("hdfs:///tmp/spark/streaming") // set checkpoint directory
        ssc
      }
  
      val ssc = StreamingContext.getOrCreate("hdfs:///tmp/spark/streaming", functionToCreateContext _)
      ssc.start()
      ssc.awaitTermination()
    }
  }