Simple Log Service:Utiliser Spark Streaming pour consommer des données de journal

Limites

La consommation via Spark Streaming n'est compatible qu'avec Spark 2.x.

Le SDK Spark d'Alibaba Cloud propose deux modes de consommation : Receiver et Direct. Ajoutez la dépendance Maven suivante à votre projet :

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

Consommer des données de journal en mode Receiver

Consommer des données de journal en mode Direct

Le mode Direct requiert un service ZooKeeper pour stocker l'état intermédiaire. Définissez un répertoire de point de contrôle dans ZooKeeper afin de persister ces données intermédiaires. Pour re-consommer les données après le redémarrage d'une tâche, supprimez le répertoire ZooKeeper correspondant et modifiez le nom du groupe de consommateurs.

• Paramètres

  Parameter	Type	Description
  project	String	Nom du projet dans SLS.
  logstore	String	Nom du logstore dans SLS.
  consumerGroup	String	Nom du groupe de consommateurs. Ce nom sert uniquement à enregistrer les points de contrôle de consommation.
  endpoint	String	Endpoint de la région où réside le projet SLS. Pour plus d'informations, consultez la page Endpoints.
  accessKeyId	String	AccessKey ID utilisé pour accéder à SLS.
  accessKeySecret	String	AccessKey Secret utilisé pour accéder à SLS.
  zkAddress	String	URL de connexion du service ZooKeeper.

• Limitation de débit

  Spark Streaming traite les données par micro-lots. Spécifiez le nombre d'entrées de journal à consommer par lot et par shard.

  Dans SLS, chaque requête d'écriture est stockée sous forme de groupe de journaux. Une requête d'écriture typique contient plusieurs entrées de journal dans un seul groupe. Lorsque le suivi web (web tracking) est activé, chaque requête d'écriture ne contient qu'une seule entrée de journal par groupe. Les paramètres suivants contrôlent le volume de données consommé lors d'un seul lot.

  Parameter	Description	Default
  spark.loghub.batchGet.step	Nombre maximal de groupes de journaux récupérés par requête de consommation.	100
  spark.streaming.loghub.maxRatePerShard	Nombre maximal d'entrées de journal consommées par shard et par lot.	10000

  Le paramètre spark.streaming.loghub.maxRatePerShard définit le nombre cible maximal d'entrées de journal par shard et par lot. Le SDK récupère les groupes de journaux par incrément de spark.loghub.batchGet.step et accumule le nombre d'entrées. Lorsque ce total atteint ou dépasse la valeur de spark.streaming.loghub.maxRatePerShard, le SDK arrête la récupération. Étant donné que la granularité de consommation se situe au niveau du groupe de journaux, la limite imposée par spark.streaming.loghub.maxRatePerShard est approximative. Le nombre réel d'entrées par lot dépend de spark.loghub.batchGet.step et du nombre d'entrées contenues dans chaque groupe de journaux.

• Exemple

  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()
    }
  }