七陌社交是一家专门做客服系统的公司, 传智教育是基于七陌社交构建客服系统，每天都有非常多的的用户进行聊天, 传智教育目前想要对这些聊天记录进行存储, 同时还需要对每天的消息量进行实时统计分析, 请您来设计如何实现数据的存储以及实时的数据统计分析工作。
需求如下:

• 1. 选择合理的存储容器进行数据存储, 并让其支持基本数据查询工作
• 2. 进行实时统计消息总量
• 3. 进行实时统计各个地区收 发 消息的总量
• 4. 进行实时统计每一位客户发送和接收消息数量

实时采集七陌用户聊天信息数据，存储消息队列Kafka，再实时将数据处理转换，将其消息存储Hudi表中，最终使用Hive和Spark业务指标统计，基于FanBI可视化报表展示。

• 1、Apache Flume：分布式实时日志数据采集框架
  由于业务端数据在不断的在往一个目录下进行生产, 我们需要实时的进行数据采集, 而flume就是一个专门用于数据采集工具,比如就可以监控某个目录下文件, 一旦有新的文件产生即可立即采集。

• 2、Apache Kafka：分布式消息队列
  Flume 采集过程中, 如果消息非常的快, Flume也会高效的将数据进行采集, 那么就需要一个能够快速承载数据容器, 而且后续还要对数据进行相关处理转换操作, 此时可以将flume采集过来的数据写入到Kafka中,进行消息数据传输，而Kafka也是整个集团中心所有业务线统一使用的消息系统, 用来对接后续的业务(离线或者实时)。

• 3、Apache Spark：分布式内存计算引擎，离线和流式数据分析处理
  整个七陌社交案例, 需要进行实时采集，那么此时也就意味着数据来一条就需要处理一条, 来一条处理一条, 此时就需要一些流式处理的框架，Structured Streaming或者Flink均可。
  此外，七陌案例中，对每日用户消息数据按照业务指标分析，最终存储MySQL数据库中，选择SparkSQL。

• 4、Apache Hudi：数据湖框架
  七陌用户聊天消息数据，最终存储到Hudi表（底层存储：HDFS分布式文件系统），统一管理数据文件，后期与Spark和Hive集成，进行业务指标分析。

• 5、Apache Hive：大数据数仓框架
  与Hudi表集成，对七陌聊天数据进行分析，直接编写SQL即可。

• 6、MySQL：关系型数据库
  将业务指标分析结果存储在MySQL数据库中，后期便于指标报表展示。

• 7、FineBI：报表工具
  帆软公司的一款商业图表工具, 让图表制作更加简单

本次案例, 直接提供专门用于生产七陌社交消息数据的工具, 可以直接部署在业务端进行数据生成即可，接下来部署用于生产数据的工具jar包。

用户聊天数据以文本格式存储日志文件中，包含20个字段，下图所示：

样本数据：

上述数据各个字段之间分割符号为：\001

运行jar包：7Mo_DataGen.jar，指定参数信息，模拟生成用户聊天信息数据，写入日志文件。

• 第一步、创建原始文件目录

mkdir -p /export/data/7mo_init

• 第二步、上传模拟数据程序

cd /export/data/7mo_init
rz

• 第三步、创建模拟数据目录

mkdir -p /export/data/7mo_data

• 第四步、运行程序生成数据

# 1. 语法
java -jar /export/data/7mo_init/7Mo_DataGen.jar 原始数据路径 模拟数据路径 随机产生数据间隔ms时间
  	
# 2. 测试：每500ms生成一条数据
java -jar /export/data/7mo_init/7Mo_DataGen.jar \
/export/data/7mo_init/7Mo_Data.xlsx \
/export/data/7mo_data \
500

• 第五步、查看产生数据
  

由于七陌用户比较多和活跃度很高，聊天信息数据比较大（每日增量：25GB至30GB），采用实时方式采集数据，此处选择框架：Apache Flume。

Aapche Flume是由Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的软件，网址：http://flume.apache.org/

Flume的核心是把数据从数据源(source)收集过来，再将收集到的数据送到指定的目的地(sink)。为了保证输送的过程一定成功，在送到目的地(sink)之前，会先缓存数据(channel),待数据真正到达目的地(sink)后，flume在删除自己缓存的数据。

当前Flume有两个版本：

• Flume 0.9X版本的统称Flume OG（original generation）
• Flume1.X版本的统称Flume NG（next generation）
  由于Flume NG经过核心组件、核心配置以及代码架构重构，与Flume OG有很大不同。改动的另一原因是将Flume纳入 apache 旗下，Cloudera Flume 改名为 Apache Flume。

Flume系统中核心的角色是agent，agent本身是一个Java进程，一般运行在日志收集节点。

每一个agent相当于一个数据传递员，内部有三个组件：

• Source：采集源，用于跟数据源对接，以获取数据；
• Sink：下沉地，采集数据的传送目的，用于往下一级agent或者往最终存储系统传递数据；
• Channel：agent内部的数据传输通道，用于从source将数据传递到sink；
  在整个数据的传输的过程中，流动的是event，它是Flume内部数据传输的最基本单元。
  

event将传输的数据进行封装，如果是文本文件，通常是一行记录，event也是事务的基本单位。event从source，流向channel，再到sink，本身为一个字节数组，并可携带headers(头信息)信息。event代表着一个数据的最小完整单元，从外部数据源来，向外部的目的地去。

一个完整的event包括：event headers、event body，其中event body是flume收集到的日记记录。

Apache Flume 的安装非常简单，直接解压，然后配置JDK环境变量即可。
第一步、上传解压

# 上传
cd /export/software
rz apache-flume-1.9.0-bin.tar.gz

# 解压，重命名及创建软链接
tar -zxf apache-flume-1.9.0-bin.tar.gz -C /export/server

cd /export/server
mv apache-flume-1.9.0-bin flume-1.9.0-bin
ln -s flume-1.9.0-bin flume

第二步、修改flume-env.sh

cd /export/server/flume/conf
mv flume-env.sh.template  flume-env.sh

vim flume-env.sh
# 22行：修改JDK路径
export JAVA_HOME=/export/server/jdk

需求说明: 监听服务器上某一个端口号(例如: 44444), 采集发向此端口的数据。

• 第1步、确定三大组件

1. source组件: 需要一个能够监听端口号的组件(网络组件)
   使用Apache Flume提供的 : NetCat TCP Source
   

2. channel组件: 需要一个传输速度更快的管道(内存组件)
   使用Apache Flume提供的 : Memory Channel

3. sink组件 : 此处我们只需要打印出来即可(日志组件)
   使用Apache Flume提供的 : Logger Sink

• 第2步、编写采集配置文件：netcat_source_logger_sink.properties

cd /export/server/flume/conf
vim netcat_source_logger_sink.properties

内容如下：

# 第一部分: 定义这个agent中各组件的名字
a1.sources = r1
a1.sinks = k1
a1.channels = c1

#第二部分:  描述和配置source组件：r1
a1.sources.r1.type = netcat
a1.sources.r1.bind = node1.itcast.cn
a1.sources.r1.port = 44444

# 第三部分: 描述和配置sink组件：k1
a1.sinks.k1.type = logger

# 第四部分: 描述和配置channel组件，此处使用是内存缓存的方式
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 第五部分: 描述和配置source  channel   sink之间的连接关系
a1.sources.r1.channels = c1 
a1.sinks.k1.channel = c1

• 第3步、启动flume: 指定采集配置文件

/export/server/flume/bin/flume-ng agent -n a1  \
-c conf -f /export/server/flume/conf/netcat_source_logger_sink.properties \
-Dflume.root.logger=INFO,console

参数说明: 	
  -c conf   指定flume自身的配置文件所在目录	
  -f conf/netcat-logger.con  指定我们所描述的采集方案	
  -n a1  指定我们这个agent的名字

• 第4步、接下来进行测试: 一定要启动之后, 连接测试
  先要往agent采集监听的端口上发送数据，让agent有数据可采。

1. 安装telnet

yum -y install telnet

2. 随便在一个能跟agent节点通信的机器上，执行如下命令

telnet node1.itcast.cn  44444

七陌社交数据源特点：持续不断的向某一个目录下得一个文件输出消息。功能要求：实时监控某一个目录下的文件, 一旦发现有新的文件,立即将其进行采集到Kafka中。

• 第1步、确定三大组件

1. source组件: 能够监控某个目录的文件source组件
   使用Apache Flume提供的 : taildir
2. channel组件: 一般都是选择 内存组件 (更高效)
   使用Apache Flume提供 : Memory Channel
3. sink组件: 输出到 Kafka的sink组件
   使用Apache Flume提供：Kafka Sink

• 第2步、编写采集配置文件：7mo_mem_kafka.properties

vim /export/server/flume/conf/7mo_mem_kafka.properties

内容如下：

# define a1
a1.sources = s1 
a1.channels = c1
a1.sinks = k1

#define s1
a1.sources.s1.type = TAILDIR
#指定一个元数据记录文件
a1.sources.s1.positionFile = /export/server/flume/position/taildir_7mo_kafka.json
#将所有需要监控的数据源变成一个组
a1.sources.s1.filegroups = f1
#指定了f1是谁：监控目录下所有文件
a1.sources.s1.filegroups.f1 = /export/data/7mo_data/.*
#指定f1采集到的数据的header中包含一个KV对
a1.sources.s1.headers.f1.type = 7mo
a1.sources.s1.fileHeader = true

#define c1
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000

#define k1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = 7MO-MSG
a1.sinks.k1.kafka.bootstrap.servers = node1.itcast.cn:9092
a1.sinks.k1.kafka.flumeBatchSize = 10
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 100

#bind
a1.sources.s1.channels = c1
a1.sinks.k1.channel = c1

• 第3步、启动ZK服务和Kafka服务

/export/server/zookeeper/bin/zkServer.sh start 
/export/server/kafka/bin/kafka-server-start.sh -daemon /export/server/kafka/config/server.properties

• 第4步、创建topic

/export/server/kafka/bin/kafka-topics.sh --create \
--topic 7MO-MSG  --partitions 3 --replication-factor 2 \
--bootstrap-server node1.itcast.cn:9092

• 第5步、启动flume: 指定采集配置文件

/export/server/flume/bin/flume-ng agent \
-n a1 -c /export/server/flume/conf/ \
-f /export/server/flume/conf/7mo_mem_kafka.properties \
-Dflume.root.logger=INFO,console

• 第6步、启动模拟数据

java -jar /export/data/7mo_init/7Mo_DataGen.jar \
/export/data/7mo_init/7Mo_Data.xlsx \
/export/data/7mo_data \
5000

观察Kafka Topic中是否有数据：

编写Spark中流式程序：StructuredStreaming，实时从Kafka消费获取社交数据，经过转换（数据字段提取等）处理，最终保存到Hudi表中，表的格式：ROM。

创建Maven Module模块，基于Spark框架编写程序，添加相关依赖，工程结构如下：

Module模块中pom.xml依赖：

<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.12.10</scala.version>
    <scala.binary.version>2.12</scala.binary.version>
    <spark.version>3.0.0</spark.version>
    <hadoop.version>2.7.3</hadoop.version>
    <hudi.version>0.9.0</hudi.version>
    <mysql.version>5.1.48</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>

    <!-- hudi-spark3 -->
    <dependency>
        <groupId>org.apache.hudi</groupId>
        <artifactId>hudi-spark3-bundle_2.12</artifactId>
        <version>${hudi.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-avro_2.12</artifactId>
        <version>${spark.version}</version>
    </dependency>
    <!-- hudi-spark3 -->
    <dependency>
        <groupId>org.apache.hudi</groupId>
        <artifactId>hudi-hive-sync</artifactId>
        <version>${hudi.version}</version>
    </dependency>

    <dependency>
        <groupId>org.apache.httpcomponents</groupId>
        <artifactId>httpcore</artifactId>
        <version>4.4.13</version>
    </dependency>
    <dependency>
        <groupId>org.apache.httpcomponents</groupId>
        <artifactId>httpclient</artifactId>
        <version>4.5.12</version>
    </dependency>

    <dependency>
        <groupId>org.lionsoul</groupId>
        <artifactId>ip2region</artifactId>
        <version>1.7.2</version>
    </dependency>

    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>${mysql.version}</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>

Hudi表数据存储在HDFS目录上， 将HDFS文件系统配置文件，放入模块Module资源目录resources下。

七陌社交数据解析封装实体类：MomoMessage ，基于Scala语言定义Case Class 样例类。

package cn.itcast.hudi.momo

/**
 * 封装Momo聊天记录实体样例类CaseClass
 */
case class MomoMessage(
                         msg_time: String,
                         sender_nickyname: String,
                         sender_account: String,
                         sender_sex: String,
                         sender_ip: String,
                         sender_os: String,
                         sender_phone_type: String,
                         sender_network: String,
                         sender_gps: String,
                         receiver_nickyname: String,
                         receiver_ip: String,
                         receiver_account: String,
                         receiver_os: String,
                         receiver_phone_type: String,
                         receiver_network: String,
                         receiver_gps: String,
                         receiver_sex: String,
                         msg_type: String,
                         distance: String,
                         message: String
                      )

后续，将Kafka消费社交数据，解析封装到实体类对象中。

创建对象object：MomoStreamHudi，编写MAIN方法，按照编写流式程序5个步骤，写出代码结构，如下所示：

package cn.itcast.hudi.momo

import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.types.StringType

/**
 * 编写StructuredStreaming流式程序：
实时消费Kafka中Momo聊天数据，进行转换处理，保存至Hudi表，并且自动同步至Hive表
 */
object MomoStreamHudi {
   
   def main(args: Array[String]): Unit = {
      // step1、构建SparkSession实例对象
      val spark: SparkSession = createSparkSession(this.getClass)
      
      // step2、从Kafka实时消费数据
      val kafkaStreamDF: DataFrame = readFromKafka(spark, "7mo-msg")
      
      // step3、提取数据，转换数据类型
      val streamDF: DataFrame = process(kafkaStreamDF)
      
      // step4、保存数据至Hudi表中：MOR（读取时保存）
      //printToConsole(streamDF)
      saveToHudi(streamDF)
      
      // step5、流式应用启动以后，等待终止
      spark.streams.active.foreach(
query => println(s"Query: ${query.name} is Running .............")
)
      spark.streams.awaitAnyTermination()
   }

}

从Spark2.x开始，程序入口SparkSession，无论SparkSQL批处理还是StructuredStreaming流计算，程序首先创建SparkSession对象，封装方法：createSparkSession

/**
 * 创建SparkSession会话实例对象，基本属性设置
 */
def createSparkSession(clazz: Class[_]): SparkSession = {
   SparkSession.builder()
      .appName(this.getClass.getSimpleName.stripSuffix("$"))
      .master("local[2]")
      // 设置序列化方式：Kryo
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      // 设置属性：Shuffle时分区数和并行度
      .config("spark.default.parallelism", 2)
      .config("spark.sql.shuffle.partitions", 2)
          .config("spark.sql.streaming.forceDeleteTempCheckpointLocation", "true")
      .getOrCreate()
}

封装方法：readFromKafka，从Kafka消费Topic数据，指定名称和Kafka Brokers地址信息。

/**
 * 指定Kafka Topic名称，实时消费数据
 */
def readFromKafka(spark: SparkSession, topicName: String): DataFrame = {
   spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "node1.itcast.cn:9092")
      .option("subscribe", topicName)
      .option("startingOffsets", "latest")
      .option("maxOffsetsPerTrigger", 100000)
      .option("failOnDataLoss", "false")
      .load()
}

流式数据打印控制台，封装方法：printToConsole，便于开发过程中测试使用。

def printToConsole(streamDF: DataFrame): Unit = {
   streamDF.writeStream
      .outputMode(OutputMode.Append())
      .queryName("query-hudi-momo")
          .format("console")
          .option("numRows", "10")
          .option("truncate", "false")
      .option("checkpointLocation", "/datas/hudi-struct-ckpt-0")
          .start()
}

对Kafka消费数据，先解析封装到实体类MomoMessage，再添加字段构建Hudi表中三大核心字段值：message_id 每条数据主键、day 分区字段及ts 数据合并字段。

/**
 * 对Kafka获取数据，进行转换操作，获取所有字段的值，转换为String，以便保存Hudi表
 */
def process(streamDF: DataFrame): DataFrame = {
   import streamDF.sparkSession.implicits._
   
   /*
      2021-11-25 20:52:58牛星海17870843110女156.35.36.204IOS 9.0华为 荣耀Play4T4G91.319474,29.033363成紫57.54.100.313946849234Android 6.0OPPO A11X4G84.696447,30.573691 女TEXT78.22KM有一种想见不敢见的伤痛，这一种爱还埋藏在我心中，让我对你的思念越来越浓，我却只能把你你放在我心中。
    */
   // 1-提取Message消息数据
   val messageStreamDF: DataFrame = streamDF.selectExpr("CAST(value AS STRING) message")
   
   // 2-解析数据，封装实体类
   val momoStreamDS: Dataset[MomoMessage] = messageStreamDF
      .as[String] // 转换为Dataset
      .map(message => {
         val array = message.split("\001")
         val momoMessage = MomoMessage(
            array(0), array(1), array(2), array(3), array(4), array(5), array(6), array(7), 
array(8), array(9),array(10), array(11), array(12), array(13), array(14), 
array(15), array(16), array(17), array(18), array(19)
         )
         // 返回实体类
         momoMessage
      })
   
   // 3-为Hudi表添加字段：主键id、数据聚合字段ts、分区字段day
   val hudiStreamDF = momoStreamDS.toDF()
      .withColumn("ts", unix_timestamp($"msg_time").cast(StringType))
      .withColumn(
         "message_id",
         concat($"sender_account", lit("_"), $"ts", lit("_"), $"receiver_account")
      )
      .withColumn("day", substring($"msg_time", 0, 10))
   
   hudiStreamDF
}

将流式数据集Stream DataFrame，使用foreachBatch方法，将每批次数据保存到Hudi表中，需要指定必要属性字段。

/**
 * 将流式数据集DataFrame保存至Hudi表，分别表类型：COW和MOR
 */
def saveToHudi(streamDF: DataFrame): Unit = {
   streamDF.writeStream
      .outputMode(OutputMode.Append())
      .queryName("query-hudi-momo")
      // 针对每微批次数据保存
      .foreachBatch((batchDF: Dataset[Row], batchId: Long) => {
         println(s"============== BatchId: $batchId start ==============")
         
         import org.apache.hudi.DataSourceWriteOptions._
         import org.apache.hudi.config.HoodieWriteConfig._
         import org.apache.hudi.keygen.constant.KeyGeneratorOptions._
         
         batchDF.write
            .format("hudi")
            .mode(SaveMode.Append)
            .option(TBL_NAME.key, "7mo_msg_hudi")
            .option(TABLE_TYPE.key(), "MERGE_ON_READ")
            .option(RECORDKEY_FIELD_NAME.key(), "message_id")
            .option(PRECOMBINE_FIELD_NAME.key(), "ts")
            .option(PARTITIONPATH_FIELD_NAME.key(), "day")
            .option(HIVE_STYLE_PARTITIONING_ENABLE.key(), "true")
            // 插入数据，产生shuffle时，分区数目
            .option("hoodie.insert.shuffle.parallelism", "2")
            .option("hoodie.upsert.shuffle.parallelism", "2")
            // 表数据存储路径
            .save("/hudi-warehouse/7mo_msg_hudi")
      })
      .option("checkpointLocation", "/datas/hudi-struct-ckpt")
      .start()
}

至此，流式程序StructuredStreaming编写完成，接下来启动各个组件服务，进行测试。

启动服务：ZK服务、Kafka服务和HDFS服务，其次运行流式应用程序，最后运行Flume Agent和模拟数据程序，查看Hudi表数据存储目录。

# NameNode和DataNode
hadoop-daemon.sh start namenode 
hadoop-daemon.sh start datanode

# ZK服务和Kafka服务
/export/server/zookeeper/bin/zkServer.sh start 
/export/server/kafka/bin/kafka-server-start.sh -daemon /export/server/kafka/config/server.properties

# Flume Agent
/export/server/flume/bin/flume-ng agent \
-c conf/ \
-n a1 \
-f /export/server/flume/conf/7mo_mem_kafka.properties \
-Dflume.root.logger=INFO,console

# 模拟数据程序
java -jar /export/data/7mo_init/7Mo_DataGen.jar \
/export/data/7mo_init/7Mo_Data.xlsx \
/export/data/7mo_data/ \
5000

Hudi存储目录结构：

至此，实时存储七陌社交数据至Hudi表，整个链路已经完成：

将Hudi表数据，与Hive表进行关联，使用beeline等客户端，编写SQL分析Hudi表数据。

启动Hive MetaStore服务和HiveServer2服务，再启动beeline客户端：

/export/server/hive/bin/start-metastore.sh
/export/server/hive/bin/start-hiveserver2.sh

/export/server/hive/bin/start-beeline.sh

编写DDL语句，创建Hive表，关联Hudi表，其中设置InputFormat实现类。

# 创建Hive表，映射到Hudi表
CREATE EXTERNAL TABLE db_hudi.tbl_7mo_hudi(
  msg_time             String,
  sender_nickyname     String,
  sender_account       String,
  sender_sex           String,
  sender_ip            String,
  sender_os            String,
  sender_phone_type    String,
  sender_network       String,
  sender_gps           String,
  receiver_nickyname   String,
  receiver_ip          String,
  receiver_account     String,
  receiver_os          String,
  receiver_phone_type  String,
  receiver_network     String,
  receiver_gps         String,
  receiver_sex         String,
  msg_type             String,
  distance             String,
  message              String,
  message_id           String,
  ts                   String       
)
PARTITIONED BY (day string)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 'org.apache.hudi.hadoop.HoodieParquetInputFormat' 
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION '/hudi-warehouse/7mo_msg_hudi' ;

由于Hudi是分区表，需要手动添加分区信息：

alter table db_hudi.tbl_7mo_hudi 
add if not exists partition(day = '2021-11-27') location '/hudi-warehouse/7mo_msg_hudi/day=2021-11-27' ;

查询Hive表前10条数据：

SELECT
  msg_time, sender_nickyname, receiver_nickyname, ts 
FROM db_hudi.tbl_7mo_hudi 
WHERE day = '2021-11-27'
limit 10 ;

编写SQL，对七陌社交数据进行简易指标统计分析，由于数据流较小，设置本地模式执行。

set hive.exec.mode.local.auto=true;
set hive.mapred.mode=nonstrict;

• 指标1：统计总消息量

WITH tmp AS (
  SELECT COUNT(1) AS momo_total  FROM db_hudi.tbl_7mo_hudi WHERE day = '2021-11-27'
)
SELECT "全国" AS momo_name, momo_total FROM tmp;

• 指标2：统计各个用户, 发送消息量

WITH tmp AS (
  SELECT 
    sender_nickyname, COUNT(1) momo_total 
  FROM db_hudi.tbl_7mo_hudi 
  WHERE day = '2021-11-27' GROUP BY sender_nickyname
)
SELECT 
  sender_nickyname AS momo_name, momo_total
FROM tmp 
ORDER BY momo_total DESC LIMIT 10;

• 指标3：统计各个用户, 接收消息量

WITH tmp AS (
  SELECT 
    receiver_nickyname, COUNT(1) momo_total 
  FROM db_hudi.tbl_7mo_hudi 
  WHERE day = '2021-11-27' GROUP BY receiver_nickyname
)
SELECT 
  receiver_nickyname AS momo_name, momo_total  
FROM tmp 
ORDER BY momo_total DESC LIMIT 10;

• 指标4：统计男女发送信息量

SELECT 
  sender_sex, receiver_sex, COUNT(1) momo_total 
FROM db_hudi.tbl_7mo_hudi 
WHERE day = '2021-11-27' GROUP BY sender_sex, receiver_sex;

编写SparkSQL程序，加载Hudi表数据封装到DataFrame中，按照业务指标需要，编写SQL分析数据，最终保存到MySQL数据库表中，流程示意图如下：

对七陌社交消息数据的实时统计操作, 如下统计需求：

• 1)、统计消息的总条数
• 2)、根据IP地址统计各个地区(省) 发送的消息数和接收的消息数
• 3)、统计七陌社交消息中各个用户发送多少条和接收多少条
  

将上述业务需求，最终结果存储到MySQL数据库1张表中：7mo.7mo_report。

其中字段：7mo_category 表示指标类型：

• 1：表示全国信息量统计
• 2：表示各省份发送信息量统计
• 3：表示各省份接收信息量统计
• 4：表示用户发送信息量统计
• 5：表示用户接收信息量统计
  在MySQL数据库，创建数据库：7mo，表：7mo_reprot，对应DDL语句如下：

-- 创建数据库
CREATE DATABASE IF NOT EXISTS 7mo ;
-- 创建表
CREATE TABLE IF NOT EXISTS `7mo`.`7mo_report` (
    `7mo_name` varchar(100) NOT NULL,
    `7mo_total` bigint(20) NOT NULL,
    `7mo_category` varchar(100) NOT NULL,
    PRIMARY KEY (`7mo_name`, `7mo_category`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 ;

创建对象object：MomoSQLHudi，编写MAIN方法，按照编写流式程序5个步骤，写出代码结构，如下所示：

package cn.itcast.hudi.momo

import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}
import org.lionsoul.ip2region.{DataBlock, DbConfig, DbSearcher}

/**
 * 编写SparkSQL程序，基于DSL和SQL分析Hudi表数据，最终保存值MySQL数据库表中
 */
object MomoSQLHudi {
   
   def main(args: Array[String]): Unit = {
      // step1、构建SparkSession实例对象
      val spark: SparkSession = createSparkSession(this.getClass)
      
      // step2、加载Hudi表数据，指定Hudi数据存储路径
      val hudiDF: DataFrame = loadHudiTable(spark, "/hudi-warehouse/7mo_msg_hudi")
      //println(s"Count = ${hudiDF.count()}")
      //hudiDF.printSchema()
      //hudiDF.show(numRows = 10, truncate = false)
      
      // step3、数据ETL转换：提取字段，解析IP为省份和城市
      val etlDF: DataFrame = etl(hudiDF)
      //println(s"Count = ${etlDF.count()}")
      //etlDF.printSchema()
      //etlDF.show(numRows = 100, truncate = false)
      
      // step4、业务指标分析
      process(etlDF)
      
      // 应用结束，关闭资源
      spark.stop()
   }
}

其中创建SparkSession对象，封装方法：createSparkSession，前面实时存储中一样。

使用Spark DataSource外部数据源接口方式，加载Hudi表数据，指定数据存储路径，封装方法：loadHudiTable。

/**
 * 指定Hudi表数据存储path，加载Hudi表数据，返回DataFrame
 */
def loadHudiTable(spark: SparkSession, tablePath: String): DataFrame = {
   val dataframe = spark.read
      .format("hudi")
      .load(tablePath)
   
   // 返回数据
   dataframe
}

解析IP地址为【省份】，推荐使用【ip2region】第三方工具库，官网网址：https://gitee.com/lionsoul/ip2region/，引入使用IP2Region第三方库：

• 第一步、复制IP数据集【ip2region.db】到工程下的【dataset】目录
  

• 第二步、在Maven中添加依赖

<dependency>
    <groupId>org.lionsoul</groupId>
    <artifactId>ip2region</artifactId>
    <version>1.7.2</version>
</dependency>

• 第三步、ip2region的使用
  
  采用自定义UDF函数方式，传递IP地址数据，解析返回Province省份：
  

除了解析IP地址为省份，还需要将业务需求中涉及到字段选择，封装方法：etl，代码如下：

/**
 * 提取字段数据和转换经纬度为省份城市
 */
def etl(dataframe: DataFrame): DataFrame = {
   val session: SparkSession = dataframe.sparkSession
   
   // 1-自定义UDF函数，解析IP地址为省份和城市
   session.udf.register(
      "ip_to_province",
      (ip: String) => {
         // 构建DbSearch对象
         val dbSearcher = new DbSearcher(new DbConfig(), "dataset/ip2region.db")
         
         // 依据IP地址解析
         val dataBlock: DataBlock = dbSearcher.btreeSearch(ip)
         // 中国|0|海南省|海口市|教育网
         val region: String = dataBlock.getRegion
         // 分割字符串，获取省份和城市
         val Array(_, _, province, _, _) = region.split("\\|")
         // 返回Region对象
         province
      }
   )
   
   // 2-提取字段和解析IP
   dataframe.createOrReplaceTempView("view_tmp_momo")
   val etlDF: DataFrame = session.sql(
      """
        |SELECT
        |  day, sender_nickyname, receiver_nickyname,
        |  ip_to_province(sender_ip) AS sender_province,
        |  ip_to_province(receiver_ip) AS receiver_province
        |FROM
        |  view_tmp_momo
        |""".stripMargin
   )
   
   // 返回结果数据
   etlDF
}

注册DataFrame为临时视图，编写SQL语句进行分析，最终将所有指标结果合并，进行保存。

/**
 * 按照业务指标分析数据
 */
def process(dataframe: DataFrame): Unit = {
   val session: SparkSession = dataframe.sparkSession
   
   // 1-将DataFrame注册为临时视图
   dataframe.createOrReplaceTempView("view_tmp_etl")
   // 2-指标1：统计总消息量
   val reportAllTotalDF: DataFrame = session.sql(
      """
        |WITH tmp AS (
        |  SELECT COUNT(1) AS 7mo_total  FROM view_tmp_etl
        |)
        |SELECT "全国" AS 7mo_name, 7mo_total, "1" AS 7mo_category FROM tmp;
        |""".stripMargin
   )
   // 2-指标2：统计各省份发送消息量
   val reportSenderProvinceTotalDF: DataFrame = session.sql(
      """
        |WITH tmp AS (
        |  SELECT sender_province, COUNT(1) AS 7mo_total FROM view_tmp_etl GROUP BY sender_province
        |)
        |SELECT sender_province AS 7mo_name, 7mo_total, "2" AS 7mo_category FROM tmp;
        |""".stripMargin
   )
   // 2-指标3：统计各省份接收消息量
   val reportReceiverProvinceTotalDF: DataFrame = session.sql(
      """
        |WITH tmp AS (
        |  SELECT receiver_province, COUNT(1) AS 7mo_total FROM view_tmp_etl GROUP BY receiver_province
        |)
        |SELECT receiver_province AS 7mo_name, 7mo_total, "3" AS 7mo_category FROM tmp;
        |""".stripMargin
   )
   // 2-指标4：统计各个用户, 发送消息量
   val reportSenderNickyNameTotalDF: DataFrame = session.sql(
      """
        |WITH tmp AS (
        |  SELECT sender_nickyname, COUNT(1) AS 7mo_total FROM view_tmp_etl GROUP BY sender_nickyname
        |)
        |SELECT sender_nickyname AS 7mo_name, 7mo_total, "4" AS 7mo_category FROM tmp;
        |""".stripMargin
   )
   // 2-指标5：统计各个用户, 接收消息量
   val reportReceiverNickyNameTotalDF: DataFrame = session.sql(
      """
        |WITH tmp AS (
        |  SELECT receiver_nickyname, COUNT(1) AS 7mo_total FROM view_tmp_etl GROUP BY receiver_nickyname
        |)
        |SELECT receiver_nickyname AS 7mo_name, 7mo_total, "5" AS 7mo_category FROM tmp;
        |""".stripMargin
   )
   // 3-保存报表至MySQL数据库
   val reportTotalDF: Dataset[Row] = reportAllTotalDF
      .union(reportSenderProvinceTotalDF)
      .union(reportReceiverProvinceTotalDF)
      .union(reportSenderNickyNameTotalDF)
      .union(reportReceiverNickyNameTotalDF)
   // reportTotalDF.show(500, truncate = false)
   reportTotalDF
      .coalesce(1)
          .write
          .mode(SaveMode.Append)
          .format("jdbc")
          .option("driver", "com.mysql.jdbc.Driver")
          .option("url", 
"jdbc:mysql://node1.itcast.cn:3306/?useUnicode=true&characterEncoding=utf-8&useSSL=false")
          .option("dbtable", "7mo.7mo_report")
          .option("user", "root")
          .option("password", "123456")
          .save()
}

其中，直接使用SparkSQL中外部数据源JDBC方式，将结果保存到MySQL数据库表中。

执行开发完成，Spark程序，加载Hudi表数据，按照业务指标计算，结果存储MySQL数据库。

• 查看MySQL数据库表数据
  

• 查询各个指标前5条数据

(SELECT 7mo_name, 7mo_total, "全国总信息量" AS "7mo.category"
FROM 7mo.7mo_report WHERE 7mo_category = 1)
UNION
(SELECT 7mo_name, 7mo_total, "省份发送信息量" AS "7mo.category"
FROM 7mo.7mo_report WHERE 7mo_category = 2 ORDER BY 7mo_total DESC LIMIT 5)
UNION
(SELECT 7mo_name, 7mo_total, "省份接收信息量" AS "7mo.category"
 FROM 7mo.7mo_report WHERE 7mo_category = 3 ORDER BY 7mo_total DESC LIMIT 5)
UNION
(SELECT 7mo_name, 7mo_total, "用户发送信息量" AS "7mo.category"
 FROM 7mo.7mo_report WHERE 7mo_category = 4 ORDER BY 7mo_total DESC LIMIT 5)
UNION
(SELECT 7mo_name, 7mo_total, "用户接收信息量" AS "7mo.category"
 FROM 7mo.7mo_report WHERE 7mo_category = 5 ORDER BY 7mo_total DESC LIMIT 5);

使用FineBI，连接数据MySQL数据库，加载业务指标报表数据，以不同图表展示。

FineBI 是帆软软件有限公司推出的一款商业智能（Business Intelligence）产品。FineBI 是定位于自助大数据分析的 BI 工具，能够帮助企业的业务人员和数据分析师，开展以问题导向的探索式分析。官网：https://www.finebi.com/

FineBI的安装：参考《FineBI Windows版安装手册》，安装完成以后，启动登录，认识基本页面。

• 启动登录
  

• 目录：首页大屏及帮助文档
  
  - 仪表盘：用于构建所有可视化报表
  

• 数据准备：用于配置各种报表的数据来源
  - 管理系统：用于管理整个FineBI的使用：用户管理、数据源管理、插件管理、权限管理等
  

创建MySQL数据库连接：【管理系统】 -> 【数据连接】 -> 【数据连接管理】

填写MySQL数据库连接信息：

数据连接名称：node1-mysql
用户名：root
密码：123456
数据连接URL：jdbc:mysql://node1.itcast.cn:3306/7mo?useUnicode=true&characterEncoding=utf8

添加MySQL数据库中业务报表：7mo_report，选择【数据准备，添加分组【七陌数据】和业务包【七陌报表】。

点进【七陌报表】，添加表，采用【SQL数据集】方式：

输入表名称和SQL语句

SELECT
  7mo_name, 7mo_total,
  CASE 7mo_category
      WHEN '1' THEN '总消息量'
      WHEN '2' THEN '各省份发送量'
      WHEN '3' THEN '各省份接收量'
      WHEN '4' THEN '各用户发送量'
      WHEN '5' THEN '各用户接收量'
  END AS 7mo_category
FROM 7mo.7mo_report 

先建仪表盘，名称为：【七陌社交数据统计报表】，如下图所示：

接下来，为仪表盘选择模板样式【预设样式5】：深蓝色海洋背景。

• 首先添加标题：【其他】 -> 【文本组件】
  

输入仪表盘名称：七陌社交数据统计报表
- 其次，添加文本组件，显示总的消息数目

选择前面添加表：7mo_report_mysql

按照下图所示：选择字段值和过滤字段类别

以柱形图方式，展示出发送信息量最多的Top10用户。

• 第1步、添加组件，选择【柱形图】，填写标题名称。
  
• 第2步、选择不同字段，设置相关过滤和显示
  
  其中，展示的数据为：用户发送信息量统计数据。
  

此外，仅仅展示Top10 发送信息量最大，需要过滤操作

展示柱形图时，按照发送信息量进行降序排序

以饼图方式，展示Top10省份发送信息量，具体操作如下所示：

• 第1步、添加组件，选择【饼图】，填写标题名称。
  

• 第2步、选择不同字段，设置相关过滤和显示
  
  其中，过滤获取各个省份发送信息量统计数据
  
  此外，仅仅获取Top10省份，发送信息量最多:
  

上述饼图中，在外边框显示数据，设置如下所示：

以地图方式，展示各省份发送信息量，具体操作如下所示：

• 第1步、添加组件，选择【区域地图】，填写标题名称。
  
• 第2步、选择省份字段映射到地理角色
  
• 第3步、选择不同字段，设置相关过滤和显示
  

其中，过滤获取各个省份发送信息量统计数据