Simple Log Service:Konsumsi Spark Streaming

Batasan

Konsumen Spark Streaming hanya mendukung versi Spark 2.x.

Mode Receiver

Mode Direct

Mode Direct memerlukan kluster ZooKeeper untuk menyimpan status antara. Anda harus menentukan path direktori checkpoint di ZooKeeper untuk data ini. Untuk mengonsumsi ulang data setelah restart pekerjaan, hapus direktori tersebut di ZooKeeper dan ubah nama kelompok konsumen.

• Parameter

  Parameter	Type	Deskripsi
  Project	String	Nama Proyek Simple Log Service.
  LogStore	String	Nama LogStore.
  consumerGroup	String	Nama kelompok konsumen, hanya digunakan untuk menyimpan titik pemeriksaan konsumsi.
  endpoint	String	Titik akhir untuk Wilayah tempat Proyek Simple Log Service berada. Untuk informasi selengkapnya, lihat Endpoints.
  accessKeyId	String	ID AccessKey yang digunakan untuk mengakses Simple Log Service.
  accessKeySecret	String	Rahasia AccessKey yang digunakan untuk mengakses Simple Log Service.
  zkAddress	String	Alamat koneksi kluster ZooKeeper Anda.

• Konfigurasi pembatasan laju

  Spark Streaming memproses data dalam mikro-batch. Untuk setiap batch, konsumen harus menentukan berapa banyak entri log yang akan diambil.

  Model penyimpanan dasar Simple Log Service menggunakan LogGroup sebagai unit. Biasanya, setiap LogGroup sesuai dengan satu permintaan tulis dan dapat berisi ribuan log. Namun, dalam skenario seperti pelacakan web, setiap permintaan tulis mungkin hanya berisi satu log. Untuk menangani skenario yang berbeda ini, SDK menyediakan dua parameter pembatasan laju:

  Parameter	Deskripsi	Default
  spark.loghub.batchGet.step	Jumlah maksimum LogGroup yang diambil dalam satu permintaan.	100
  spark.streaming.loghub.maxRatePerShard	Jumlah maksimum entri log yang dikonsumsi dari satu shard per batch.	10000

  Parameter spark.streaming.loghub.maxRatePerShard menetapkan target jumlah maksimum entri log yang diambil dari satu shard dalam setiap batch. SDK mengambil LogGroup secara bertahap sesuai dengan nilai spark.loghub.batchGet.step hingga jumlah total log mencapai atau melebihi target yang ditetapkan oleh spark.streaming.loghub.maxRatePerShard. Karena itu, spark.streaming.loghub.maxRatePerShard berfungsi sebagai batas lunak. Jumlah aktual log yang diambil per batch akan bervariasi tergantung pada spark.loghub.batchGet.step dan jumlah log dalam setiap LogGroup.

• Contoh

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

Untuk informasi selengkapnya, lihat proyek di GitHub.