spark读写数据仓库

1、使用场景

 随着业务及数据量的增长，数据库中的数据大致可以分为两类，一类为操作型数据，另一类为分析型数据。其中，操作型数据通常与日常业务紧密相关且可进行增删改查，而分析型数据通常为历史数据，用于统计分析，仅能查询不可增删改。此外，分析型数据有时需要对业务数据进行数据清洗得到。因此，可以将分析型数据导入数据仓库hive中，spark再定时从hive中取出数据进行分析。以城市空气质量预测为例，空气监测点分布在城市中的各个地方，定时地将数据上传至平台中，为了对城市空气质量进行预测，需定期将城市中各监测点的小时数据取平均值后存入hive中，spark再定期从hive中取出数据进行预测分析。

2、spark存入hive

 spark存入hive表有两种方式，一种调用方式DF.write.saveAsTable，另一种方式调用hiveContext.sql将数据导入hive中。首先，spark从数据库中读取原始数据并进行数据清洗，求出城市中所有点的平均值代码如下：

    mpInfoList = spark.read.format("com.mongodb.spark.sql.DefaultSource") \
        .option("spark.mongodb.input.uri", MONITOR_POINT_INFO_URL) \
        .option("pipeline", matchCity) \
        .load().select("ID").rdd.map(lambda x: x.ID).collect()
    print(mpInfoList)
    airQualityData = spark.read.format("com.mongodb.spark.sql.DefaultSource")\
        .option("spark.mongodb.input.uri", INPUT_URL)\
        .load()
    airQualityData = airQualityData.filter(airQualityData.NodeIdentifier.isin(mpInfoList))

    airQualityData = airQualityData.groupBy('ComponentTime')\
        .avg('pm25', 'temp', 'press', 'humi', 'wind_speed', 'wind_dir')\
        .orderBy(airQualityData.ComponentTime)\
        .withColumnRenamed('avg(pm25)', 'pm25')\
        .withColumnRenamed('avg(temp)', 'temp')\
        .withColumnRenamed('avg(press)', 'press')\
        .withColumnRenamed('avg(humi)', 'humi')\
        .withColumnRenamed('avg(wind_speed)', 'wind_speed')\
        .withColumnRenamed('avg(wind_dir)', 'wind_dir')

随后，将清洗后的数据存入hive中，代码如下：

data.write.saveAsTable("test.airData", None, "overwrite", None)

3、spark从hive中读取数据

调用sparkSession.sql从hive中读取数据，代码如下：

    data = spark.sql("select * from test.airData")
    data1 = data.orderBy(data.ComponentTime)