4.4 Rate Source

以固定的速率生成固定格式的数据, 用来测试 Structured Streaming 的性能.

package com.atguigu.ss

import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * Author lzc
  * Date 2019/8/13 11:42 AM
  */
object RateSourceDemo {
    def main(args: Array[String]): Unit = {
        val spark: SparkSession = SparkSession
            .builder()
            .master("local[*]")
            .appName("RateSourceDemo")
            .getOrCreate()

        val rows: DataFrame = spark.readStream
            .format("rate") // 设置数据源为 rate
            .option("rowsPerSecond", 10) // 设置每秒产生的数据的条数, 默认是 1
            .option("rampUpTime", 1) // 设置多少秒到达指定速率 默认为 0
            .option("numPartitions", 2) /// 设置分区数  默认是 spark 的默认并行度
            .load

        rows.writeStream
            .outputMode("append")
            .trigger(Trigger.Continuous(1000))
            .format("console")
            .start()
            .awaitTermination()
    }
}