选择2014-2020年中连续3年的全国气象数据，对数据做数据清洗操作，如果该气象站已停用，则过滤掉该站点数据，对过滤后的数据中空值进行填补，mysql数据库建表，将清洗完毕的气象数据集导入hbase/mysql数据库中，按照需求文档对数据进行不同维度的数据分析（时间（小时，天，月，年），地点（气象站，城市，省），指标类型（指标列表如下： AQI PM2.5_24h PM10_24h SO2_24h NO2_24h O3_24h CO ）等，将结果存入mysql数据库。

好的，这是一个比较复杂的问题，需要分几个步骤来完成。我会尽可能详细地回答你的问题。

1. 数据获取和清洗

首先，你需要从官方网站或其他可靠的数据源获取2014-2020年中连续3年的全国气象数据。然后，你需要使用Python编写脚本来对数据进行清洗和整理，以使其符合数据库的格式。

具体来说，你需要遵循以下步骤：

- 读取原始数据文件，例如CSV文件。
- 过滤掉停用的气象站点的数据。
- 将缺失值替换为合适的值，例如平均值或中位数。
- 将数据格式化为数据库表格的格式。
- 将清洗后的数据保存到新的文件或内存中。

2. 数据库建表

在将清洗完毕的气象数据导入数据库之前，你需要创建一个数据库表来存储气象数据。你可以选择使用MySQL或HBase来存储数据，这里我会分别介绍如何在这两种数据库中建立表格。

在MySQL中，你可以使用以下SQL语句来创建气象数据表：

CREATE TABLE weather (
    id INT AUTO_INCREMENT PRIMARY KEY,
    station VARCHAR(50),
    city VARCHAR(50),
    province VARCHAR(50),
    date DATE,
    hour INT,
    aqi INT,
    pm25 FLOAT,
    pm10 FLOAT,
    so2 FLOAT,
    no2 FLOAT,
    o3 FLOAT,
    co FLOAT
);

在HBase中，你需要使用HBase Shell或编写Java代码来创建表格。以下是使用HBase Shell创建表格的示例：

create 'weather', {NAME => 'data'}

3. 数据导入数据库

在创建表格之后，你可以使用Python中的数据库API（如MySQLdb、psycopg2、happybase等）连接到数据库，并将数据导入数据库中。以下是一个基本的Python代码示例，用于将气象数据导入MySQL数据库中：

import MySQLdb

# 打开数据库连接
db = MySQLdb.connect("localhost","testuser","test123","weather" )

# 使用cursor()方法获取操作游标 
cursor = db.cursor()

# 将数据插入表中
data = [("station1", "city1", "province1", "2021-01-01", 1, 10, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0),
        ("station2", "city2", "province2", "2021-01-01", 1, 20, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0),
        ("station3", "city3", "province3", "2021-01-01", 1, 30, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0)]

for d in data:
    sql = "INSERT INTO weather(station, city, province, date, hour, aqi, pm25, pm10, so2, no2, o3, co) VALUES ('%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s')" % (d[0], d[1], d[2], d[3], d[4], d[5], d[6], d[7], d[8], d[9], d[10], d[11])
    try:
        cursor.execute(sql)
        db.commit()
    except:
        db.rollback()

# 关闭数据库连接
db.close()

在HBase中，你需要使用HBase API或编写Java代码来将数据导入数据库中。以下是使用happybase库将数据导入HBase数据库的示例：

import happybase

# 连接到HBase数据库
connection = happybase.Connection('localhost')
table = connection.table('weather')

# 将数据插入表中
data = [("station1", "city1", "province1", "2021-01-01", 1, 10, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0),
        ("station2", "city2", "province2", "2021-01-01", 1, 20, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0),
        ("station3", "city3", "province3", "2021-01-01", 1, 30, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0)]

for d in data:
    row_key = d[0] + d[3] + str(d[4])
    table.put(row_key, {'data:station': d[0],
                        'data:city': d[1],
                        'data:province': d[2],
                        'data:date': d[3],
                        'data:hour': str(d[4]),
                        'data:aqi': str(d[5]),
                        'data:pm25': str(d[6]),
                        'data:pm10': str(d[7]),
                        'data:so2': str(d[8]),
                        'data:no2': str(d[9]),
                        'data:o3': str(d[10]),
                        'data:co': str(d[11])})

# 关闭数据库连接
connection.close()

4. 数据分析

在将数据导入数据库之后，你可以使用SQL或HBase API进行数据分析。具体来说，你可以按照需求文档中的不同维度（时间、地点、指标类型）对数据进行分析，并将结果存储到MySQL数据库中。

以下是一个示例SQL查询，用于按小时和AQI指标对数据进行聚合：

SELECT hour, AVG(aqi) AS avg_aqi
FROM weather
GROUP BY hour

如果你使用HBase，则需要编写Java代码来查询数据。以下是一个示例Java代码，用于按小时和AQI指标对数据进行聚合：

Scan scan = new Scan();
scan.addColumn(Bytes.toBytes("data"), Bytes.toBytes("hour"));
scan.addColumn(Bytes.toBytes("data"), Bytes.toBytes("aqi"));

AggregationClient aggregationClient = new AggregationClient(configuration);
long rowCount = aggregationClient.rowCount(TableName.valueOf("weather"), new LongColumnInterpreter(), scan);
System.out.println("Total row count: " + rowCount);

Aggregation aggregation = new LongSumAggregation();
Scan scan = new Scan();
scan.addColumn(Bytes.toBytes("data"), Bytes.toBytes("hour"));
scan.addColumn(Bytes.toBytes("data"), Bytes.toBytes("aqi"));

Map<byte[], Long> result = aggregationClient.rowCount(TableName.valueOf("weather"), new LongColumnInterpreter(), scan);
for (Map.Entry<byte[], Long> entry : result.entrySet()) {
    System.out.println(Bytes.toString(entry.getKey()) + ": " + entry.getValue());
}

aggregationClient.close();

希望这可以帮助你解决问题。如果你还有其他问题，请随时问我。