客快物流大数据项目(五十四)：初始化Spark流式计算程序

package cn.it.logistics.etl.realtime

import cn.it.logistics.common.Configuration
import org.apache.commons.lang.SystemUtils
import org.apache.spark.SparkConf
import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 测试消费kafka的数据
 * 1）物流相关的数据
 * 2）客户关系管理系统的数据
 */
object App {
  /**
   * 入口函数
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {
    /**
     * 实现步骤：
     * 1）初始化spark的运行环境
     * 2）判断当前的运行环境（local/linux运行环境）
     * 3）创建sparkSession对象
     * 4）初始化物流topic数据的连接参数
     * 5）初始化客户关系系统topic数据的连接参数
     * 6）消费oracle->ogg->kafka的topic数据
     * 7）消费mysql->canal->kafka的topic数据
     * 8）启动运行等待停止
     */
    //1）初始化spark的运行环境
    val conf: SparkConf = new SparkConf()
      //设置应用的名称
      .set("spark.app.name", this.getClass.getSimpleName)
      //设置时区
      .set("spark.sql.session.timeZone", "Asia/Shanghai")
      //设置单个分区可容纳的最大字节数，默认是128M， 等同于block块的大小
      .set("spark.sql.files.maxPartitionBytes", "134217728")
      //设置合并小文件的阈值，避免每个小文件占用一个分区的情况
      .set("spark.sql.files.openCostInBytes", "134217728")
      //设置join或者shuffle的时候使用的分区数，默认情况下分区数是200
      .set("spark.sql.shuffle.partitions", "600")
      //设置join操作时可以广播到worker节点的最大字节大小，可以避免shuffer操作
      .set("spark.sql.autoBroadcastJoinThreshold", "67108864")

    //2）判断当前的运行环境（local/linux运行环境）
    if (SystemUtils.IS_OS_WINDOWS || SystemUtils.IS_OS_MAC) {
      //本地环境LOCAL_HADOOP_HOME
      System.setProperty("hadoop.home.dir", Configuration.LOCAL_HADOOP_HOME)
      //设置运行环境和checkpoint路径
      conf.set("spark.master", "local[*]").set("spark.sql.streaming.checkpointLocation", Configuration.sparkAppWinCheckpointDir)
    } else {
      //生产环境
      conf.set("spark.master", "yarn").set("spark.sql.streaming.checkpointLocation", Configuration.sparkAppDfsCheckpointDir)
    }

    //3）创建sparkSession对象
    val sparkSession: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    sparkSession.sparkContext.setLogLevel(Configuration.LOG_OFF)

    //4）初始化物流topic数据的连接参数
    val logisticsKafkaParams: Map[String, String] = Map[String, String](
      "kafka.bootstrap.servers" -> Configuration.kafkaAddress,
      "subscribe" -> Configuration.kafkaLogisticsTopic,
      "group.id" -> "logistics",
      //表示数据丢失以后（topic被删除，或者offset不存在可用的范围的时候）
      "failOnDataLoss" -> "false"
    )

    //5）初始化客户关系系统topic数据的连接参数
    val crmKafkaParams: Map[String, String] = Map[String, String](
      "kafka.bootstrap.servers" -> Configuration.kafkaAddress,
      "subscribe" -> Configuration.kafkaCrmTopic,
      "group.id" -> "logistics",
      //表示数据丢失以后（topic被删除，或者offset不存在可用的范围的时候）
      "failOnDataLoss" -> "false"
    )

    //导入隐士转换
    import sparkSession.implicits._

    //6）消费oracle->ogg->kafka的topic数据
    val logisticsDF: DataFrame = sparkSession.readStream.format("kafka").options(logisticsKafkaParams).load().selectExpr("CAST(value AS STRING)").as[String].toDF()

    //7）消费mysql->canal->kafka的topic数据
    val crmDF: DataFrame = sparkSession.readStream.format("kafka").options(crmKafkaParams).load().selectExpr("CAST(value AS STRING)").as[String].toDF()

    //输出数据
   logisticsDF.writeStream.outputMode(OutputMode.Update()).format("console").queryName("logistics").start()
    crmDF.writeStream.outputMode(OutputMode.Update()).format("console").queryName("crm").start()

    //8）启动运行等待停止
    val stream = sparkSession.streams
    //stream.active：获取当前活动流式查询的列表
    stream.active.foreach(query => println(s"准备启动的查询：${query.name}"))
    //线程阻塞，等待终止
    stream.awaitAnyTermination()
  }
}

.set("spark.sql.session.timeZone", "Asia/Shanghai")

.set("spark.sql.files.maxPartitionBytes", "134217728")

.set("spark.sql.files.openCostInBytes", "134217728")

.set("spark.sql.shuffle.partitions", "600")

.set("spark.sql.autoBroadcastJoinThreshold", "67108864")