Spark 数据统计并将结果转存至MSSQL

在 使用Spark读取Hive中的数据 中，我们演示了如何使用python编写脚本，提交到spark，读取并输出了Hive中的数据。在实际应用中，在读取完数据后，通常需要使用pyspark中的API来对数据进行统计或运算，并将结果保存起来。本节将演示这一过程。

环境准备

Hive建表并填充测试数据

本文假设你已经安装、配置好了HDFS、Hive和Spark，在Hive中创建了数据仓库Eshop，在其下创建了OrderInfo表，基于Retailer和Year进行了分区，并填充了以下数据（注意Retailer和Year是虚拟列）：

实际上，这篇文章的orderinfo表是基于上一篇 Hive中分区和分桶的概念和操作 进行构建的，因此建议先阅读一下。

安装MSSQL的JDBC驱动程序

在本文中，需要将运算的结果转存至MS Sql Server数据库，而要通过java连接MSSQL，需要在服务器上安装jdbc驱动。首先下载驱动，地址是：下载 Microsoft SQL Server JDBC 驱动程序

按下图选择sqljdbc_7.0.0.0_chs.tar.gz压缩包，然后点击“Next”下载：

解压缩之后，将根目录下的mssql-jdbc-7.0.0.jre8.jar文件，拷贝到Spark服务器上的$SPARK_HOME/jars文件夹下。

如果是搭建了一个Spark集群，那么务必将该文件拷贝至集群内所有节点的 $SPARK_HOME/jars 文件夹下。

从Windows拷贝文件到Linux有很多种方法，可以通过FTP上传，也可以通过pscp直接从Windows上拷贝至Linux，参见：免密码从windows复制文件到linux。

MSSql建表StatOrderInfo

假设要统计的是每年每个经销商的订单总数（OrderCount）、销售总额（TotalAmount）、用户数（CustomerCount），那么可以这样建表：

USE StatEShop
GO
CREATE TABLE [dbo].[Stat_OrderInfo](
    [Id] [int] IDENTITY(1,1) NOT NULL,
    [Year] [int] NOT NULL,
    [Retailer] [varchar](50) NOT NULL,
    [OrderCount] [int] NOT NULL,
    [CustomerCount] [int] NOT NULL,
    [TotalAmount] [money] NOT NULL
    CONSTRAINT [PK_stat_orderinfo] PRIMARY KEY CLUSTERED 
(
    [Id] ASC
)
) ON [PRIMARY]

需要注意订单总数和用户总数的区别：用户总数是去除重复后的下单数，即同一个用户下了10个订单，订单数为10，但是用户数为1。

编写python脚本

在向Spark提交任务作业时，可以采用三种语言的脚本，Scala、Java和Python，因为Python相对而言比较轻量（脚本语言），比较好学，因此我选择了使用Python。大多数情况下，使用哪种语言并没有区别，但在Spark SQL中，Python不支持DataSet，仅支持DataFrame，而Java和Scala则两种类型都支持。DataSet相对DataFrame的优势就是取行数据时是强类型的，而在其他方面DataSet和DataFrame的API都是相似的。

下面是本次任务的python脚本，位于D:\python\dataclean\eshop\stat_orderinfo.py：

from pyspark.sql import SparkSession
from pyspark.sql import HiveContext
from pyspark.sql import functions as F

spark = SparkSession.builder.master("spark://node0:7077")\
    .appName("eshop.year.retailer")\
    .config("spark.sql.warehouse.dir", "/user/hive/warehouse")\
    .config("hive.metastore.uris", "thrift://192.168.1.56:9083")\
    .enableHiveSupport()\
    .getOrCreate()

hiveCtx = HiveContext(spark)
df = hiveCtx.sql("select * from eshop.orderinfo")

df2 = df.groupBy("year", "retailer").agg(
    F.count("*").alias("OrderCount"),
    F.sum("OrderAmount").alias("TotalAmount"),
    F.countDistinct("Customer").alias("CustomerCount")
)

options = {
    "url": "jdbc:sqlserver://192.168.1.103:1433;databaseName=StatEShop",
    "user":"sa",
    "password":"your db password",
    "driver":"com.microsoft.sqlserver.jdbc.SQLServerDriver"
}

df2.write.format("jdbc").options(dbtable="Stat_OrderInfo", **options)\
    .mode("append")\
    .save()

本例中的数据统计逻辑很简单，如果要学习spark都可以执行哪些运算，请参考官方的文档：pyspark.sql module。这个文档需要花大量时间认真学习一下，才能对Spark的操作有更深入的了解。

上面的代码有几下几点还需要注意一下：

• 这里我是运行在Spark集群上，其中的master节点是node0，因此是这样创建spark对象的：spark = SparkSession.builder.master("spark://node0:7077")。如果是本地运行，则将spark://node0:7077替换为local
• Hive的metasotre服务需要先运行，也就是要已经执行过：hive --service metastore。具体参见：使用Spark读取Hive中的数据
• F.sum("OrderAmount").alias("TotalAmount") 语句用于改名，否则，聚合函数执行完毕后，列名为 sum(OrderAmount)。

先在Windows上执行下面的命令，将stat_orderinfo.py拷贝至Linux的/root/python/eshop目录：

# pscp -i D:\linux\keys\awwork.ppk D:\python\dataclean\eshop\stat_orderinfo.py root@192.168.1.56:/root/python/eshop

然后在配置好Spark的服务器上执行：

# $SPARK_HOME/bin/spark-submit /root/python/eshop/stat_orderinfo.py

执行过程中如果一切正常将不会有任何输出，此时，如果访问 http://node0:8080，可以看到spark作业正在执行：

node0是Spark集群的主节点，地址是一个局域网地址：192.168.1.56。

点击 application ID，会进入到作业的执行明细中，注意此时浏览器地址变为了 http://node0:4040。

4040端口号只有在作业执行阶段可以访问，而因为我们的数据量很少，运算逻辑也极为简单，因此这个作业通常10几秒就执行完成了。当作业执行完成后，这个页面也就无法访问了。

打开SQL Server管理器，可以看到下面的结果：

Select * from stat_orderinfo;
Id    Year     Retailer       OrderCount  CustomerCount TotalAmount
----- -------- -------------- ----------  ------------- ------------
1     2017     Apple          2           2             8380.00
2     2018     XiaoMi         3           2             2740.00
3     2017     XiaoMi         1           1             2010.00
4     2018     Apple          1           1             920.00

至此，已经成功完成了Spark数据统计并转存到MSSQL Server的作业任务。

感谢阅读，希望这篇文章能给你带来帮助！