---

前言

每年天猫双十一购物节，都会有一块巨大的实时作战大屏，展现当前的销售情况。这种炫酷的页面背后，其实有着非常强大的技术支撑，而这种场景其实就是实时报表分析。

---

1、业务需求概述

​ 模拟交易订单数据，发送至分布式消息队列Kafka，实时消费交易订单数据进行分析处理，业务流程图如下所示：

实时从Kafka消费交易订单数据，按照不同维度实时统计【销售订单额】，最终报表Report结果存储MySQL数据库；

二 项目代码

1.模拟交易数据

编写程序，实时产生交易订单数据，使用Json4J类库转换数据为JSON字符，发送Kafka Topic中，代码如下：

// =================================== 订单实体类 =================================
package cn.itcast.spark.mock

/**
 * 订单实体类（Case Class）
 * @param orderId     订单ID
 * @param userId      用户ID
 * @param orderTime   订单日期时间
 * @param ip          下单IP地址
 * @param orderMoney  订单金额
 * @param orderStatus 订单状态
 */
case class OrderRecord(
                          orderId: String,
                          userId: String,
                          orderTime: String,
                          ip: String,
                          orderMoney: Double,
                          orderStatus: Int
                      )


// ================================== 模拟订单数据 ==================================
package cn.itcast.spark.mock

import java.util.Properties

import org.apache.commons.lang3.time.FastDateFormat
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer
import org.json4s.jackson.Json

import scala.util.Random

/**
 * 模拟生产订单数据，发送到Kafka Topic中
 *     Topic中每条数据Message类型为String，以JSON格式数据发送
 * 数据转换：
 *     将Order类实例对象转换为JSON格式字符串数据（可以使用json4s类库）
 */
object MockOrderProducer {
    
    def main(args: Array[String]): Unit = {
        
        var producer: KafkaProducer[String, String] = null
        try {
            // 1. Kafka Client Producer 配置信息
            val props = new Properties()
            props.put("bootstrap.servers", "node1.itcast.cn:9092")
            props.put("acks", "1")
            props.put("retries", "3")
            props.put("key.serializer", classOf[StringSerializer].getName)
            props.put("value.serializer", classOf[StringSerializer].getName)
            
            // 2. 创建KafkaProducer对象，传入配置信息
            producer = new KafkaProducer[String, String](props)
            
            // 随机数实例对象
            val random: Random = new Random()
            // 订单状态：订单打开 0，订单取消 1，订单关闭 2，订单完成 3
            val allStatus =Array(0, 1, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
            
            while(true){
                // 每次循环 模拟产生的订单数目
                val batchNumber: Int = random.nextInt(1) + 5
                (1 to batchNumber).foreach{number =>
                    val currentTime: Long = System.currentTimeMillis()
                    val orderId: String = s"${getDate(currentTime)}%06d".format(number)
                    val userId: String = s"${1 + random.nextInt(5)}%08d".format(random.nextInt(1000))
                    val orderTime: String = getDate(currentTime, format="yyyy-MM-dd HH:mm:ss.SSS")
                    val orderMoney: String = s"${5 + random.nextInt(500)}.%02d".format(random.nextInt(100))
                    val orderStatus: Int = allStatus(random.nextInt(allStatus.length))
                    // 3. 订单记录数据
                    val orderRecord: OrderRecord = OrderRecord(
                        orderId, userId, orderTime, getRandomIp, orderMoney.toDouble, orderStatus
                    )
                    // 转换为JSON格式数据
                    val orderJson = new Json(org.json4s.DefaultFormats).write(orderRecord)
                    println(orderJson)
                    // 4. 构建ProducerRecord对象
                    val record = new ProducerRecord[String, String]("orderTopic", orderJson)
                    // 5. 发送数据：def send(messages: KeyedMessage[K,V]*), 将数据发送到Topic
                    producer.send(record)
                }
                Thread.sleep(random.nextInt(100) + 500)
            }
        }catch {
            case e: Exception => e.printStackTrace()
        }finally {
            if(null != producer) producer.close()
        }
    }
    
    /**=================获取当前时间=================*/
    def getDate(time: Long, format: String = "yyyyMMddHHmmssSSS"): String = {
        val fastFormat: FastDateFormat = FastDateFormat.getInstance(format)
        val formatDate: String = fastFormat.format(time)  // 格式化日期
        formatDate
    }
    
    /**================= 获取随机IP地址 =================*/
    def getRandomIp: String = {
        // ip范围
        val range: Array[(Int, Int)] = Array(
            (607649792,608174079), //36.56.0.0-36.63.255.255
            (1038614528,1039007743), //61.232.0.0-61.237.255.255
            (1783627776,1784676351), //106.80.0.0-106.95.255.255
            (2035023872,2035154943), //121.76.0.0-121.77.255.255
            (2078801920,2079064063), //123.232.0.0-123.235.255.255
            (-1950089216,-1948778497),//139.196.0.0-139.215.255.255
            (-1425539072,-1425014785),//171.8.0.0-171.15.255.255
            (-1236271104,-1235419137),//182.80.0.0-182.92.255.255
            (-770113536,-768606209),//210.25.0.0-210.47.255.255
            (-569376768,-564133889) //222.16.0.0-222.95.255.255
        )
        // 随机数：IP地址范围下标
        val random = new Random()
        val index = random.nextInt(10)
        val ipNumber: Int = range(index)._1 + random.nextInt(range(index)._2 - range(index)._1)

        // 转换Int类型IP地址为IPv4格式
        number2IpString(ipNumber)
    }
    
    /**=================将Int类型IPv4地址转换为字符串类型=================*/
    def number2IpString(ip: Int): String = {
        val buffer: Array[Int] = new Array[Int](4)
        buffer(0) = (ip >> 24) & 0xff
        buffer(1) = (ip >> 16) & 0xff
        buffer(2) = (ip >> 8) & 0xff
        buffer(3) = ip & 0xff
        // 返回IPv4地址
        buffer.mkString(".")
    }
    
}

2.创建Maven模块

创建Maven模块，加入相关依赖，具体内如如下：

    <repositories>
        <repository>
            <id>aliyun</id>
            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
        </repository>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
        <repository>
            <id>jboss</id>
            <url>http://repository.jboss.com/nexus/content/groups/public</url>
        </repository>
    </repositories>

    <properties>
        <scala.version>2.11.12</scala.version>
        <scala.binary.version>2.11</scala.binary.version>
        <spark.version>2.4.5</spark.version>
        <hadoop.version>2.6.0-cdh5.16.2</hadoop.version>
        <kafka.version>2.0.0</kafka.version>
        <mysql.version>8.0.19</mysql.version>
    </properties>

    <dependencies>

        <!-- 依赖Scala语言 -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>
        <!-- Spark Core 依赖 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <!-- Spark SQL 依赖 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <!-- Structured Streaming + Kafka  依赖 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql-kafka-0-10_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <!-- Hadoop Client 依赖 -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <!-- Kafka Client 依赖 -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.0.0</version>
        </dependency>
        <!-- 根据ip转换为省市区 -->
        <dependency>
            <groupId>org.lionsoul</groupId>
            <artifactId>ip2region</artifactId>
            <version>1.7.2</version>
        </dependency>
        <!-- MySQL Client 依赖 -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>${mysql.version}</version>
        </dependency>
        <!-- JSON解析库：fastjson -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.47</version>
        </dependency>

    </dependencies>

    <build>
        <outputDirectory>target/classes</outputDirectory>
        <testOutputDirectory>target/test-classes</testOutputDirectory>
        <resources>
            <resource>
                <directory>${project.basedir}/src/main/resources</directory>
            </resource>
        </resources>
        <!-- Maven 编译的插件 -->
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.0</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

项目结构如下:

3.核心代码

RealTimeOrderReport.java

package cn.itcast.spark.report

import java.util.concurrent.TimeUnit

import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.{OutputMode, Trigger}
import org.apache.spark.sql.expressions.{UserDefinedAggregateFunction, UserDefinedFunction}
import org.apache.spark.sql.types.{DataType, DataTypes}
import org.lionsoul.ip2region.{DataBlock, DbConfig, DbSearcher}

	def printToConsole(dataFrame: DataFrame) = {
		dataFrame.writeStream
  		.format("console")
  		.outputMode(OutputMode.Update())
  		.option("numRows","50")
  		.option("truncate","false")
  		.start()
	}



	def main(args: Array[String]): Unit = {
		//1.获取spark实例对象
		val spark: SparkSession = SparkSession.builder()
			.appName("isDemo")
			.master("local[3]")
			.config("spark.sql.shuffle.partitions", "3")
			.getOrCreate()
		import spark.implicits._

		val dataFrame: DataFrame = spark.readStream
			.format("kafka")
			.option("kafka.bootstrap.servers", "node1.itcast.cn:9092")
			.option("subscribe", "orderTopic")
			.load()
			.selectExpr("CAST (value AS STRING)")
//		printToConsole(dataFrame)


val ip_to_region: UserDefinedFunction = udf((ip: String) => {
	// 1. 创建DbSearch对象，指定数据字典文件位置
	val dbSearcher = new DbSearcher(new DbConfig(), "src/main/dataset/ip2region.db")
	// 2. 传递IP地址，解析获取数据
	val dataBlock: DataBlock = dbSearcher.btreeSearch(ip)
	// 3. 获取解析省份和城市
	val region: String = dataBlock.getRegion
	//println(region)  // 中国|0|海南省|海口市|教育网
	val Array(_, _, pronvice, city, _) = region.split("\\|")
	(pronvice, city)
})
		val frame: DataFrame = dataFrame
			.select(
				get_json_object($"value", "$.ip").as("ip"),
				get_json_object($"value", "$.orderMoney")
					.cast(DataTypes.createDecimalType(10, 2))
					.as("money"),
				get_json_object($"value", "$.orderStatus").as("status")
			)
			.filter($"status" === 0)
			.withColumn("region", ip_to_region($"ip"))
			.select(
				$"region._1".as("province"),
				$"region._2".as("city"),
				$"money"
			)


//				printToConsole(frame)

//		/**
//			* 订单实体类（Case Class）
//			* @param orderId     订单ID
//			* @param userId      用户ID
//			* @param orderTime   订单日期时间
//			* @param ip          下单IP地址
//			* @param orderMoney  订单金额
//			* @param orderStatus 订单状态
//			*/
//

//		printToConsole(dframe)
		//SELECT "国家" as type, SUM(money) as totalMoney   FROM tmp_view
		//SELECT province as type, SUM(money) as totalMoney   FROM tmp_view GROUP BY province
		//SELECT city as type, SUM(money) as totalMoney   FROM (SELECT * FROM tmp_view WHERE city in ("北京市", "上海市", "深圳市", "广州市", "杭州市", "成都市", "南京市", "武汉市", "西安市"))t GROUP BY t.city
		frame.createOrReplaceTempView("tmp_view")
		val f: DataFrame = spark.sql(
			"""
				|SELECT "国家" as type, SUM(money) as totalMoney   FROM tmp_view
  """.stripMargin)
		val f2: DataFrame = spark.sql(
			"""
				|SELECT province as type, SUM(money) as totalMoney   FROM tmp_view GROUP BY province
  """.stripMargin)
		val f3: DataFrame = spark.sql(
			"""
				|SELECT city as type, SUM(money) as totalMoney   FROM (SELECT * FROM tmp_view WHERE city in ("北京市", "上海市", "深圳市", "广州市", "杭州市", "成都市", "南京市", "武汉市", "西安市"))t GROUP BY t.city
  """.stripMargin)
//		printToConsole(f3)
		saveToMySQL(f,"total")
		saveToMySQL(f2,"totalprovince")
		saveToMySQL(f3,"totalcity")


		spark.streams.awaitAnyTermination()
	}
	def saveToMySQL(streamDF:DataFrame,reportType:String)={
		streamDF.writeStream
  		.outputMode(OutputMode.Complete())
  		.queryName(s"${reportType}")
  		.foreachBatch((batchDF:DataFrame,batchId:Long)=>{
				batchDF.coalesce(1)
  				.write.mode(SaveMode.Overwrite)
  				.format("jdbc")
  				.option("url","jdbc:mysql://node1.itcast.cn:3306/?serverTimezone=UTC&characterEncoding=utf8&useUnicode=true")
  				.option("driver","com.mysql.cj.jdbc.Driver")
  				.option("user","root")
  				.option("password","123456")
  				.option("dbtable",s"db_spark.tb_order${reportType}")
  				.save()
			}
			)
			.option("checkpointLocation", s"datas/spark/structured-ckpt-${System.currentTimeMillis()}")
			.start()
	}
}

OrderRecord.scala

package cn.itcast.spark.mock

/**
 * 订单实体类（Case Class）
 * @param orderId     订单ID
 * @param userId      用户ID
 * @param orderTime   订单日期时间
 * @param ip          下单IP地址
 * @param orderMoney  订单金额
 * @param orderStatus 订单状态
 */
case class OrderRecord(
                          orderId: String,
                          userId: String,
                          orderTime: String,
                          ip: String,
                          orderMoney: Double,
                          orderStatus: Int
                      )

总结

总结：
​ 实时报表分析是近年来很多公司采用的报表统计方案之一，其中最主要的应用就是实时大屏展示。利用流式计算实时得出结果直接被推送到前端应用，实时显示出重要指标的变换情况。

​ 最典型的案例便是淘宝双十一活动，每年双十一购物节，除疯狂购物外，最引人注目的就是双十一大屏不停跳跃的成交总额。在整个计算链路中包括从天猫交易下单购买到数据采集，数据计算，数据校验，最终落到双十一大屏上展示的全链路时间压缩在5秒以内，顶峰计算性能高达数三十万笔订单/秒，通过多条链路流计算备份确保万无一失。

	这次的双十一实时报表分析实战主要用SQL编写,尚未用DSL编写,这是有待完善的地方.
	此次的天猫双十一实时报表分享就到这里,喜欢的小伙伴欢迎一键三连!!