基于零售交易数据的Spark数据处理与分析

【版权声明】版权所有，严禁转载，严禁用于商业用途，侵权必究。
作者：厦门大学信息学院计算机科学系2019级研究生何昕
指导老师：厦门大学数据库实验室林子雨博士/副教授
相关教材：林子雨、郑海山、赖永炫编著《Spark编程基础（Python版）》（访问教材官网）
相关案例：基于Python语言的Spark数据处理分析案例集锦（PySpark）

本案例数据集是来自Kaggle的一个跨国在线零售业务的交易数据，采用Python为编程语言，采用Hadoop存储数据，采用Spark对数据进行处理分析，并使用Echarts做数据可视化。

一、环境搭建

本次作业使用的环境和软件如下：
（1）Linux操作系统：Ubuntu 16.04
（2）Python：3.5.2
（3）Hadoop：3.1.3 （查看安装教程）
（4）Spark：2.4.0 （查看安装教程）
（5）Bottle：v0.12.18
Bottle是一个快速、简洁、轻量级的基于WSIG的微型Web框架，此框架除了Python的标准库外，不依赖任何其他模块。安装方法是，打开Linux终端，执行如下命令：

sudo apt-get install python3-pip
pip3 install bottle

至此，环境搭建过程结束。

二、数据预处理

本次作业使用的数据集是来自Kaggle的一个跨国在线零售业务的交易数据（从百度网盘下载数据集，提取码：pyej），该公司在英国注册，主要销售礼品。数据集E_Commerce_Data.csv包含541909个记录，时间跨度为2010-12-01到2011-12-09，每个记录由8个属性组成，具体的含义如下表：
字段名称类型含义举例
InvoiceNo string 订单编号（退货订单以C开头） 536365
StockCode string 产品代码 85123A
Description string 产品描述 WHITE METAL LANTERN
Quantity integer 购买数量（负数表示退货） 6
InvoiceDate string 订单日期和时间 12/1/2010 8:26
UnitPrice double 单价（英镑） 3.39
CustomerID integer 客户编号 17850
Country string 国家名称 United Kingdom
首先，将数据集E_Commerce_Data.csv上传至hdfs上，命令如下：

hdfs dfs -put E_Commerce_Data.csv

接着，使用如下命令进入pyspark的交互式编程环境，对数据进行初步探索和清洗：

cd /usr/local/spark #进入Spark安装目录
./bin/pyspark

（1）读取在HDFS上的文件，以csv的格式读取，得到DataFrame对象。

>>> df=spark.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('E_Commerce_Data.csv')

（2）查看数据集的大小，输出541909，不包含标题行

>>> df.count()

（3）打印数据集的schema，查看字段及其类型信息。输出内容就是上文中的属性表。

>>> df.printSchema()

（4）创建临时视图data。

>>> df.createOrReplaceTempView("data")

（5）由于顾客编号CustomID和商品描述Description均存在部分缺失，所以进行数据清洗，过滤掉有缺失值的记录。特别地，由于CustomID为integer类型，所以该字段若为空，则在读取时被解析为0，故用df["CustomerID"]!=0 条件过滤。

>>> clean=df.filter(df["CustomerID"]!=0).filter(df["Description"]!="")

（6）查看清洗后的数据集的大小，输出406829。

>>> clean.count()

（7）数据清洗结束。根据作业要求，预处理后需要将数据写入HDFS。将清洗后的文件以csv的格式，写入E_Commerce_Data_Clean.csv中（实际上这是目录名，真正的文件在该目录下，文件名类似于part-00000），需要确保HDFS中不存在这个目录，否则写入时会报“already exists”错误。

>>> clean.write.format("com.databricks.spark.csv").options(header='true',inferschema='true').save('E_Commerce_Data_Clean.csv')

至此，数据预处理完成。接下来将使用python编写应用程序，对清洗后的数据集进行统计分析。

三、数据分析

备注：全部源代码文件可以从百度网盘下载：（从百度网盘下载源代码，提取码：pyej）
创建project.py文件，用于编写应用程序。
首先，导入需要用到的python模块。

# -*- coding: utf-8 -*-
from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.sql.types import StringType, DoubleType, IntegerType, StructField, StructType
import json
import os

接着，获取spark sql的上下文。

sc = SparkContext('local', 'spark_project')
sc.setLogLevel('WARN')
spark = SparkSession.builder.getOrCreate()

最后，从HDFS中以csv的格式读取清洗后的数据目录E_Commerce_Data_Clean.csv，程序会取出该目录下的所有数据文件，得到DataFrame对象，并创建临时视图data用于后续分析。

df = spark.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('E_Commerce_Data_Clean.csv')
df.createOrReplaceTempView("data")

为方便统计结果的可视化，将结果导出为json文件供web页面渲染。使用save方法导出数据：

def save(path, data):
    with open(path, 'w') as f:
        f.write(data)

准备工作完成。接下来对数据进行分析，分为概览和关系两个部分。

1.概览

（1）客户数最多的10个国家

每个客户由编号CustomerID唯一标识，所以客户的数量为COUNT(DISTINCT CustomerID)，再按照国家Country分组统计，根据客户数降序排序，筛选出10个客户数最多的国家。得到的countryCustomerDF为DataFrame类型，执行collect()方法即可将结果以数组的格式返回。

def countryCustomer():
    countryCustomerDF = spark.sql("SELECT Country,COUNT(DISTINCT CustomerID) AS countOfCustomer FROM data GROUP BY Country ORDER BY countOfCustomer DESC LIMIT 10")
    return countryCustomerDF.collect()

最后调用save方法就可以将结果导出至文件了，格式如下：

[国家名称,客户数]

（2）销量最高的10个国家

Quantity字段表示销量，因为退货的记录中此字段为负数，所以使用SUM(Quantity)即可统计出总销量，即使有退货的情况。再按照国家Country分组统计，根据销量降序排序，筛选出10个销量最高的国家。得到的countryQuantityDF为DataFrame类型，执行collect()方法即可将结果以数组的格式返回。

def countryQuantity():
    countryQuantityDF = spark.sql("SELECT Country,SUM(Quantity) AS sumOfQuantity FROM data GROUP BY Country ORDER BY sumOfQuantity DESC LIMIT 10")
    return countryQuantityDF.collect()

最后调用save方法就可以将结果导出至文件了，格式如下：

[国家名称,销量]

（3）各个国家的总销售额分布情况

UnitPrice 字段表示单价，Quantity字段表示销量，退货的记录中Quantity字段为负数，所以使用SUM(UnitPrice*Quantity)即可统计出总销售额，即使有退货的情况。再按照国家Country分组统计，计算出各个国家的总销售额。得到的countrySumOfPriceDF为DataFrame类型，执行collect()方法即可将结果以数组的格式返回。

def countrySumOfPrice():
    countrySumOfPriceDF = spark.sql("SELECT Country,SUM(UnitPrice*Quantity) AS sumOfPrice FROM data GROUP BY Country")
    return countrySumOfPriceDF.collect()

最后调用save方法就可以将结果导出至文件了，格式如下：

[国家名称,总销售额]

（4）销量最高的10个商品

Quantity字段表示销量，退货的记录中Quantity字段为负数，所以使用SUM(Quantity)即可统计出总销量，即使有退货的情况。再按照商品编码StockCode分组统计，计算出各个商品的销量。得到的stockQuantityDF为DataFrame类型，执行collect()方法即可将结果以数组的格式返回。

def stockQuantity():
    stockQuantityDF = spark.sql("SELECT StockCode,SUM(Quantity) AS sumOfQuantity FROM data GROUP BY StockCode ORDER BY sumOfQuantity DESC LIMIT 10")
    return stockQuantityDF.collect()

最后调用save方法就可以将结果导出至文件了，格式如下：

[商品编号,销量]

（5）商品描述的热门关键词Top300

Description字段表示商品描述，由若干个单词组成，使用LOWER(Description)将单词统一转换为小写。此时的结果为DataFrame类型，转化为rdd后进行词频统计，再根据单词出现的次数进行降序排序，流程图如下：

得到的结果为RDD类型，为其制作表头wordCountSchema，包含word和count属性，分别为string类型和integer类型。调用createDataFrame()方法将其转换为DataFrame类型的wordCountDF，将word为空字符串的记录剔除掉，调用take()方法得到出现次数最多的300个关键词，以数组的格式返回。

def wordCount():
    wordCount = spark.sql("SELECT LOWER(Description) as description from data").rdd.flatMap(lambda line:line['description'].split(" ")).map(lambda word:(word,1)).reduceByKey(lambda a,b:a+b).repartition(1).sortBy(lambda x:x[1],False)
 
    wordCountSchema = StructType([StructField("word", StringType(), True),StructField("count", IntegerType(), True)])
    wordCountDF = spark.createDataFrame(wordCount, wordCountSchema)
    wordCountDF = wordCountDF.filter(wordCountDF["word"]!='')
    return wordCountDF.take(300)

最后调用save方法就可以将结果导出至文件了，格式如下：

[关键词,次数]

（6）退货订单数最多的10个国家

InvoiceNo字段表示订单编号，所以订单总数为COUNT(DISTINCT InvoiceNo)，由于退货订单的编号的首个字母为C，例如C540250，所以利用WHERE InvoiceNo LIKE 'C%'子句即可筛选出退货的订单，再按照国家Country分组统计，根据退货订单数降序排序，筛选出10个退货订单数最多的国家。得到的countryReturnInvoiceDF为DataFrame类型，执行collect()方法即可将结果以数组的格式返回。

def countryReturnInvoice():
    countryReturnInvoiceDF = spark.sql("SELECT Country,COUNT(DISTINCT InvoiceNo) AS countOfReturnInvoice FROM data WHERE InvoiceNo LIKE 'C%' GROUP BY Country ORDER BY countOfReturnInvoice DESC LIMIT 10")
    return countryReturnInvoiceDF.collect()

最后调用save方法就可以将结果导出至文件了，格式如下：

[国家名称,退货订单数]

2.关系

（7）月销售额随时间的变化趋势

统计月销售额需要3个字段的信息，分别为订单日期InvoiceDate，销量Quantity和单价UnitPrice。由于InvoiceDate字段格式不容易处理，例如“8/5/2011 16:19”，所以需要对这个字段进行格式化操作。由于统计不涉及小时和分钟数，所以只截取年月日部分，并且当数值小于10时补前置0来统一格式，期望得到年、月、日3个独立字段。先实现formatData()方法，利用rdd对日期、销量和单价字段进行处理。

def formatData():
    tradeRDD = df.select("InvoiceDate","Quantity","UnitPrice",).rdd
    result1 = tradeRDD.map(lambda line: (line['InvoiceDate'].split(" ")[0], line['Quantity'] , line['UnitPrice']))
    result2 = result1.map(lambda line: (line[0].split("/"), line[1], line[2]))
    result3 = result2.map(lambda line: (line[0][2], line[0][0] if len(line[0][0])==2 else "0"+line[0][0], line[0][1] if len(line[0][1])==2 else "0"+line[0][1], line[1], line[2]))
    return result3

流程图如下：

由于要统计的是月销售额的变化趋势，所以只需将日期转换为“2011-08”这样的格式即可。而销售额表示为单价乘以销量，需要注意的是，退货时的销量为负数，所以对结果求和可以表示销售额。RDD的转换流程如下：

得到的结果为RDD类型，为其制作表头schema，包含date和tradePrice属性，分别为string类型和double类型。调用createDataFrame()方法将其转换为DataFrame类型的tradePriceDF，调用collect()方法将结果以数组的格式返回。

def tradePrice():
    result3 = formatData()
    result4 = result3.map(lambda line:(line[0]+"-"+line[1],line[3]*line[4]))
    result5 = result4.reduceByKey(lambda a,b:a+b).sortByKey()
    schema = StructType([StructField("date", StringType(), True),StructField("tradePrice", DoubleType(), True)])
    tradePriceDF = spark.createDataFrame(result5, schema)
    return tradePriceDF.collect()

最后调用save方法就可以将结果导出至文件了，格式如下：

[日期,销售额]

（8）日销量随时间的变化趋势

由于要统计的是日销量的变化趋势，所以只需将日期转换为“2011-08-05”这样的格式即可。先调用上例的formatData()方法对日期格式进行格式化。RDD的转换流程如下：

得到的结果为RDD类型，为其制作表头schema，包含date和saleQuantity属性，分别为string类型和integer类型。调用createDataFrame()方法将其转换为DataFrame类型的saleQuantityDF，调用collect()方法将结果以数组的格式返回。

def saleQuantity():
    result3 = formatData()
    result4 = result3.map(lambda line:(line[0]+"-"+line[1]+"-"+line[2],line[3]))
    result5 = result4.reduceByKey(lambda a,b:a+b).sortByKey()
    schema = StructType([StructField("date", StringType(), True),StructField("saleQuantity", IntegerType(), True)])
    saleQuantityDF = spark.createDataFrame(result5, schema)
    return saleQuantityDF.collect()

最后调用save方法就可以将结果导出至文件了，格式如下：

[日期,销量]

（9）各国的购买订单量和退货订单量的关系

InvoiceNo字段表示订单编号，退货订单的编号的首个字母为C，例如C540250。利用COUNT(DISTINCT InvoiceNo)子句统计订单总量，再分别用WHERE InvoiceNo LIKE 'C%'和WHERE InvoiceNo NOT LIKE 'C%'统计出退货订单量和购买订单量。接着按照国家Country分组统计，得到的returnDF和buyDF均为DataFrame类型，分别表示退货订单和购买订单，如下所示：

再对这两个DataFrame执行join操作，连接条件为国家Country相同，得到一个DataFrame。但是这个DataFrame中有4个属性，包含2个重复的国家Country属性和1个退货订单量和1个购买订单量，为减少冗余，对结果筛选3个字段形成buyReturnDF。如下所示：

最后执行collect()方法即可将结果以数组的格式返回。

def buyReturn():
    returnDF = spark.sql("SELECT Country AS Country,COUNT(DISTINCT InvoiceNo) AS countOfReturn FROM data WHERE InvoiceNo LIKE 'C%' GROUP BY Country")
    buyDF = spark.sql("SELECT Country AS Country2,COUNT(DISTINCT InvoiceNo) AS countOfBuy FROM data WHERE InvoiceNo NOT LIKE 'C%' GROUP BY Country2")
    buyReturnDF = returnDF.join(buyDF, returnDF["Country"] == buyDF["Country2"], "left_outer")
    buyReturnDF = buyReturnDF.select(buyReturnDF["Country"],buyReturnDF["countOfBuy"],buyReturnDF["countOfReturn"])
    return buyReturnDF.collect()

最后调用save方法就可以将结果导出至文件了，格式如下：

[国家名称,购买订单数,退货订单数]

（10）商品的平均单价与销量的关系

由于商品的单价UnitPrice是不断变化的，所以使用平均单价AVG(DISTINCT UnitPrice)来衡量一个商品。再利用SUM(Quantity)计算出销量，将结果按照商品的编号进行分组统计，执行collect()方法即可将结果以数组的格式返回。

def unitPriceSales():
    unitPriceSalesDF = spark.sql("SELECT StockCode,AVG(DISTINCT UnitPrice) AS avgUnitPrice,SUM(Quantity) AS sumOfQuantity FROM data GROUP BY StockCode")
    return unitPriceSalesDF.collect()

最后调用save方法就可以将结果导出至文件了，格式如下：

[商品编号,平均单价,销量]

3.小结

在project.py中添加main函数，将上面的分析过程整合起来方便进行调用，代码如下：

if __name__ == "__main__":
 
    base = "static/"
    if not os.path.exists(base):
        os.mkdir(base)
 
    m = {
        "countryCustomer": {
            "method": countryCustomer,
            "path": "countryCustomer.json"
        },
        "countryQuantity": {
            "method": countryQuantity,
            "path": "countryQuantity.json"
        },
        "countrySumOfPrice": {
            "method": countrySumOfPrice,
            "path": "countrySumOfPrice.json"
        },
        "stockQuantity": {
            "method": stockQuantity,
            "path": "stockQuantity.json"
        },
        "wordCount": {
            "method": wordCount,
            "path": "wordCount.json"
        },
        "countryReturnInvoice": {
            "method": countryReturnInvoice,
            "path": "countryReturnInvoice.json"
        },
        "tradePrice": {
            "method": tradePrice,
            "path": "tradePrice.json"
        },
        "saleQuantity": {
            "method": saleQuantity,
            "path": "saleQuantity.json"
        },
        "buyReturn": {
            "method": buyReturn,
            "path": "buyReturn.json"
        },
        "unitPriceSales": {
            "method": unitPriceSales,
            "path": "unitPriceSales.json"
        }
    }
 
    for k in m:
        p = m[k]
        f = p["method"]
        save(base + m[k]["path"], json.dumps(f()))
        print ("done -> " + k + " , save to -> " + base + m[k]["path"])

上面的代码将所有的函数整合在变量 m中，通过循环调用上述所有方法并导出json文件到当前路径的static目录下。
最后利用如下指令运行分析程序：

cd /usr/local/spark
./bin/spark-submit project.py

四、可视化方法

备注：全部源代码文件可以从百度网盘下载：（从百度网盘下载源代码，提取码：pyej）
本次作业的结果可视化使用百度开源的免费数据展示框架Echarts。Echarts是一个纯Javascript的图表库，可以流畅地运行在PC和移动设备上，兼容当前绝大部分浏览器，底层依赖轻量级的Canvas类库ZRender，提供直观，生动，可交互，可高度个性化定制的数据可视化图表。
编写web.py程序，实现一个简单的web服务器，代码如下：

from bottle import route, run, static_file
import json
 
@route('/static/<filename>')
def server_static(filename):
    return static_file(filename, root="./static")
 
@route("/<name:re:.*\.html>")
def server_page(name):
    return static_file(name, root=".")
 
@route("/")
def index():
    return static_file("index.html", root=".")
 
run(host="0.0.0.0", port=9999)

bottle服务器对接收到的请求进行路由，规则如下：
（1）访问/static/时，返回静态文件
（2）访问/.html时，返回网页文件
（3）访问/时，返回首页index.html
服务器的9999端口监听来自任意ip的请求（前提是请求方能访问到这台服务器）。
首页index.html的主要代码如下(由于篇幅较大，只截取主要的部分)

<!DOCTYPE html>
<html lang="en">
 
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width,height=device-height">
    <title>E-Commerce-Data 在线零售业务数据分析</title>
<style>
/* 省略 */
    </style>
</head>
 
<body>
    <div class="container">
    /* 只展示第一个统计结果的代码，其余省略 */
        <div class="chart-group">
            <h3>(1) 客户数最多的10个国家
            <br>
            <small style="font-size: 72%;">
                ——英国的客户最多，达到3950个，数量远大于其他国家；其次是德国、法国、西班牙等
            </small>
            </h3>
            <iframe src="countryCustomer.html" class="frame" frameborder="0"></iframe>
        </div>
    </div>
    <script>document.body.clientHeight;</script>
</body>
</html>

图表页通过一个iframe嵌入到首页中。以第一个统计结果的网页countryCustomer.html为例，展示主要代码：

<!DOCTYPE html>
<html lang="en">
 
<head>
    <meta charset="UTF-8">
    <style>
/* 省略 */
    </style>
</head>
 
<body>
<div id="chart" style="width:95%;height:95%;"></div>
<script src="static/jquery-3.2.1.min.js"></script>
<script src="static/echarts-4.7.0.min.js"></script>
<script>
    var myChart = echarts.init(document.getElementById('chart'));
    myChart.setOption(
    {
        color: ['#3398DB'],
        tooltip: {
            trigger: 'axis',
            axisPointer: {
                type: 'shadow'
            }
        },
        grid: {
            left: '3%',
            right: '4%',
            bottom: '3%',
            containLabel: true
        },
        xAxis: [
            {
                name: '国家',
                data: [],
                axisTick: {
                    alignWithLabel: true
                },
                axisLabel: {  
                   interval:0,  
                   rotate:40  
                }
            }
        ],
        yAxis: [
            {
                name: '客户数',
            }
        ],
        series: [
            {
                name: '客户数',
                type: 'bar',
                barWidth: '60%',
                data: []
            }
        ]
    });
    myChart.showLoading();
    $.getJSON("/static/countryCustomer.json", data => {
        var names=[];
        var nums=[];
 
        data = data.map(v => ({
            country: v[0],
            customer: parseInt(v[1]),
        }))
 
        for(var i=0;i<data.length;i++){
            names.push(data[i].country);
            nums.push(data[i].customer);
        }
        myChart.setOption({
            xAxis: {
                data: names
            },
            series: [{
                data: nums
            }]
        });
        myChart.hideLoading();
    })
</script>
</body>
</html>

代码完成后，在代码所在的根目录下执行以下指令启动web服务器：

python3 web.py

若打印出以下信息则表示web服务启动成功。接着，可以通过使用浏览器访问网页的方式查看统计结果。

Bottle v0.12.18 server starting up (using WSGIRefServer())...
Listening on http://0.0.0.0:9999/
Hit Ctrl-C to quit.

为方便运行程序，编写run.sh脚本，内容如下。首先向spark提交project.py程序对数据进行统计分析，生成的json文件会存入当前路径的static目录下；接着运行web.py程序，即启动web服务器对分析程序生成的json文件进行解析渲染，方便用户通过浏览器查看统计结果的可视化界面。

#!/bin/bash
cd /usr/local/spark
./bin/spark-submit project.py
python3 web.py

提示：运行程序前，应先按照本作业的步骤二，将清洗后的数据集保存到HDFS分布式文件系统中，目录命名为E_Commerce_Data_Clean.csv，否则project.py程序会由于读取不到其中的数据文件而报错。

五、结果可视化

1.概览

（1）客户数最多的10个国家

——英国的客户最多，达到3950个，数量远大于其他国家；其次是德国、法国、西班牙等。

（2）销量最高的10个国家

——英国的销量最高，达到4008533件，远大于其他国家；其次是新西兰、爱尔兰、德国等。

（3）各个国家的总销售额分布情况

——英国的总销售额最高，达到6767873.394英镑，占比81.54%

（4）销量最高的10个商品

——编号为84077的商品销量最高，达到53215件；销量Top3的商品在数量上差距并不大。

（5）商品描述的热门关键词Top300

——热门关键词包括bag、red、heart、pink、christmas、cake等。

（6）退货订单数最多的10个国家

——英国的退货订单最多，达到3208个，远大于其他国家；其次是德国、法国、爱尔兰等。

2.关系

（7）月销售额随时间的变化趋势

——销售额较高的月份主要集中在下半年；由于该公司主要售卖礼品，并且下半年的节日较多，所以销售额比上半年高；2011年12月的销售额较低是因为数据只统计到2011/12/9 。

（8）日销量随时间的变化趋势

——下半年的日销量整体上高于上半年；2011年10月5号达到日销量的最高纪录45741件。

（9）各国的购买订单量和退货订单量的关系

——购买订单量越大的国家，退货订单量往往也越大。

（10）商品的平均单价与销量的关系

——总体上看，商品的销量随着平均单价的升高而下降。

六、总结

在完成本次作业的过程中，我综合应用了多个技术，包括利用分布式文件系统HDFS读写数据集；使用pyspark交互式编程环境对数据进行预处理；利用python语言编写应用程序，使用Spark Core的RDD编程和Spark SQL编程实现对数据的统计分析；最后使用Bottle启动web服务，结合Echarts可视化框架完成统计结果的展示。通过本次作业，我对大数据技术有了更深的理解，对大数据背景下的数据处理与分析的流程有了更清晰的认识，动手能力也得到了很大的提高。回顾这一学期的学习，我收获颇丰。十分感谢老师的授课视频、教材和PPT，这种一站式的资源极大地提升了我学习的效率。

厦大数据库实验室博客

一、环境搭建

二、数据预处理

三、数据分析

1.概览

（1）客户数最多的10个国家

（2）销量最高的10个国家

（3）各个国家的总销售额分布情况

（4）销量最高的10个商品

（5）商品描述的热门关键词Top300

（6）退货订单数最多的10个国家

2.关系

（7）月销售额随时间的变化趋势

（8）日销量随时间的变化趋势

（9）各国的购买订单量和退货订单量的关系

（10）商品的平均单价与销量的关系

3.小结

四、可视化方法

五、结果可视化

1.概览

（1）客户数最多的10个国家

（2）销量最高的10个国家

（3）各个国家的总销售额分布情况

（4）销量最高的10个商品

（5）商品描述的热门关键词Top300

（6）退货订单数最多的10个国家

2.关系

（7）月销售额随时间的变化趋势

（8）日销量随时间的变化趋势

（9）各国的购买订单量和退货订单量的关系

（10）商品的平均单价与销量的关系

六、总结