基于Hadoop实现气象分析大屏可视化项目【源码+LW+PPT+解析】

作者简介：Java领域优质创作者、CSDN博客专家、CSDN内容合伙人、掘金特邀作者、阿里云博客专家、51CTO特邀作者、多年架构师设计经验、多年校企合作经验，被多个学校常年聘为校外企业导师，指导学生毕业设计并参与学生毕业答辩指导，有较为丰富的相关经验。期待与各位高校教师、企业讲师以及同行交流合作

主要内容：Java项目、Python项目、前端项目、PHP、ASP.NET、人工智能与大数据、单片机开发、物联网设计与开发设计、简历模板、学习资料、面试题库、技术互助、就业指导等

业务范围：免费功能设计、开题报告、任务书、中期检查PPT、系统功能实现、代码编写、论文编写和辅导、论文降重、长期答辩答疑辅导、腾讯会议一对一专业讲解辅导答辩、模拟答辩演练、和理解代码逻辑思路等。

收藏点赞不迷路关注作者有好处

文末获取源码

项目编号：BS-BD-023

一，环境介绍

语言环境：Python3.8+Hadoop

数据库：Mysql: mysql5.7

WEB框架：Django

开发工具：IDEA或PyCharm

二，项目简介

信息化社会内需要与之针对性的信息获取途径，但是途径的扩展基本上为人们所努力的方向，由于站在的角度存在偏差，人们经常能够获得不同类型信息，这也是技术最为难以攻克的课题。针对气象分析大屏可视化等问题，对气象进行研究分析，然后开发设计出气象分析大屏可视化系统以解决问题。

气象分析大屏可视化系统主要功能模块包括后台首页，系统用户（管理员），模块管理（日照时数，平均相对湿度，年降水量，平均气温，贵阳气象分析，气温对比），采取面对对象的开发模式进行软件的开发和硬体的架设，能很好的满足实际使用的需求，完善了对应的软体架设以及程序编码的工作，采取MySQL作为后台数据的主要存储单元，采用Hadoop框架、python技术、Ajax技术进行业务系统的编码及其开发，实现了本系统的全部功能。本次报告，首先分析了研究的背景、作用、意义，为研究工作的合理性打下了基础。针对气象分析大屏可视化系统的各项需求以及技术问题进行分析，证明了系统的必要性和技术可行性，然后对设计系统需要使用的技术软件以及设计思想做了基本的介绍，最后来实现气象分析大屏可视化系统和部署运行使用它。

三，系统展示

四，核心代码展示

import json
import csv
import ast
import os
from flask import jsonify
from core.mysql import MysqlPool
from service import service_selectmysqlPool = MysqlPool()# 帮助方法,合并对象
def obj_update(*config):config_temp = {}for o in config:config_temp.update(o)return config_temp# 权限集合
dict_auth = {}# 控制器父类
class Controller:def __init__(self, config):"""构造函数@param {Dictionary} config 配置参数"""# 配置参数self.config = config or {}# 添加服务self.service = service_select(self.config["services"])cg = {# 选择的模板那路径模板"tpl":"./index/",# 选择的服务"services":"user",# 注册get请求路由"get": ["list", "view", "table"],# 注册post请求路由"post": [],# 注册get api路由"get_api": ["del","get_list","get_obj","count","count_group","sum","sum_group","avg","avg_group","list_group","bar_group","get_hits_list","get_business_order_list"],# 注册post api路由"post_api": ["add", "del", "set", "import_db", "export_db", "upload"],"interact": [],"unique": []}if config:if "interact" in config:config["interact"].extend(cg["interact"])else:config["interact"] = cg["interact"]if "get" in config:config["get"].extend(cg["get"])else:config["get"] = cg["get"]if "post" in config:config["post"].extend(cg["post"])else:config["post"] = cg["post"]if "get_api" in config:config["get_api"].extend(cg["get_api"])else:config["get_api"] = cg["get_api"]if "post_api" in config:config["post_api"].extend(cg["post_api"])else:config["post_api"] = cg["post_api"]if "unique" in config:config["unique"].extend(cg["unique"])else:config["unique"] = cg["unique"]# 公共模型,用于在render（）为传递模板数据补充def model(self, ctx, model):m = {}m.update(model)## model_temp.user = ctx.session.user# 获取导航service = service_select("nav")m["nav_top"] = service.Get_list({"location": "top"})m["nav_side"] = service.Get_list({"location": "side"})m["nav_foot"] = service.Get_list({"location": "foot"})# 获取轮播图service = service_select("slides")m["list_slides"] = service.Get_list({})# 获取公告service = service_select("notice")m["list_notice"] = service.Get_list({},{"orderby": "`update_time` desc"})# 交互模型接口if ("interact" in self.config) and self.config["interact"]:m = self.model_interact(ctx, m)m["query"] = ctx.querym["body"] = ctx.bodym["auth"] = ctx.authreturn m# 交互对象def interact_obj(self, ctx, o):interact = self.config["interact"]if interact:source_table = service_select(self.config["services"]).config["table"]source_field = source_table + "_id"# 评论if "comment" in interact:service = service_select("comment")source_id = o[source_field]o["comment_list"] = service.Get_list({"source_table": source_table,"source_field": source_field,"source_id": source_id}, {"page": 1,"size": 10,})o["comment_len"] = service.Count({"source_table": source_table,"source_field": source_field,"source_id": source_id,})# 评分if "score" in interact:service = service_select("score")source_id = o[source_field]o["score_list"] = service.Get_list({"source_table": source_table,"source_field": source_field,"source_id": source_id}, {"page": 1,"size": 10,})o["score_len"] = service.Avg({"source_table": source_table,"source_field": source_field,"source_id": source_id,},{"field": "score_num"},)# 收藏if "collect" in interact:service = service_select("collect")source_id = o[source_field]o["collect_list"] = service.Get_list({"source_table": source_table,"source_field": source_field,"source_id": source_id}, {"page": 1,"size": 10,})o["collect_len"] = service.Count({"source_table": source_table,"source_field": source_field,"source_id": source_id,})# 点赞if "praise" in interact:service = service_select("praise")source_id = o[source_field]o["praise_list"] = service.Get_list({"source_table": source_table,"source_field": source_field,"source_id": source_id,}, {"page": 1,"size": 10,})o["praise_len"] = service.Count({"source_table": source_table,"source_field": source_field,"source_id": source_id,})return o# 交互列表def interact_list(self, ctx, list_1):interact = self.config["interact"]if interact:source_table = service_select(self.config["services"]).config["table"]source_field = source_table + "_id"# 评论数if "comment" in interact:service = service_select("comment")for o in list_1:source_id = o[source_field]o["comment_len"] = service.Count({"source_table": source_table,"source_field": source_field,"source_id": source_id,})# 平均分if "score" in interact:service = service_select("score")for o in list_1:source_id = o[source_field]o["score_len"] = service.Avg({"source_table": source_table,"source_field": source_field,"source_id": source_id,},{"field": "score_num"},)# 收藏人数if "collect" in interact:service = service_select("collect")for o in list_1:source_id = o[source_field]o["collect_len"] = service.Count({"source_table": source_table,"source_field": source_field,"source_id": source_id,})# 点赞人数if "praise" in interact:service = service_select("praise")for o in list_1:source_id = o[source_field]o["praise_len"] = service.Count({"source_table": source_table,"source_field": source_field,"source_id": source_id,})# 交互模型def model_interact(self, ctx, m):if ("list" in m) and m["list"]:self.interact_list(ctx, m["list"])elif ("obj" in m) and m["obj"]:self.interact_obj(ctx, m["obj"])return m"""公共参数校验"""def Check_param(self, ctx):return True# 首页def Index(self, ctx):"""首页@param {Object} ctx http请求上下文@return {Object} 返回html页面"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldif "page" in query:page = query.pop("page")config_plus["page"] = pageif "size" in query:size = query.pop("size")config_plus["size"] = sizeresult_list = self.service.Get_list(query, obj_update(self.config, config_plus))result_dict = {"list": result_list}model = self.model(ctx, result_dict)return ctx.render(self.config["tpl"] + "index" + ".html", model)def Api(self, ctx):return {"demo": "hello world!"}#  列表页面def List(self, ctx):"""列表页面@param {Object} ctx http请求上下文@return {Object} 返回html页面"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldif "page" in query:page = query.pop("page")config_plus["page"] = pageif "size" in query:size = query.pop("size")config_plus["size"] = sizeresult_list = self.service.Get_list(query, obj_update(self.config, config_plus))result_dict = {"list": result_list}model = self.model(ctx, result_dict)return ctx.render(self.config["tpl"] + "list" + ".html", model)# 表格页面def Table(self, ctx):"""表格页面@param {Object} ctx http请求上下文@return {Object} 返回html页面"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldif "page" in query:page = query.pop("page")config_plus["page"] = pageif "size" in query:size = query.pop("size")config_plus["size"] = sizeresult_list = self.service.Get_list(query, obj_update(self.config, config_plus))result_dict = {"list": result_list}model = self.model(ctx, result_dict)return ctx.render(self.config["tpl"] + "table" + ".html", model)# 详情页面def View(self, ctx):"""详情页面@param {Object} ctx http请求上下文@return {Object} 返回html页面"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldobj_result = self.service.Get_obj(query,obj_update(self.config, config_plus))obj_dict = {"obj": obj_result}model = self.model(ctx, obj_dict)return ctx.render(self.config["tpl"] + "view" + ".html", model)# 编辑页面def Edit(self, ctx):"""编辑页面@param {Object} ctx http请求上下文@return {Object} 返回html页面"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldobj_result = self.service.Get_obj(query,obj_update(self.config, config_plus))obj_dict = {"obj": obj_result}model = self.model(ctx, obj_dict)return ctx.render(self.config["tpl"] + "edit" + ".html", model)# 增def Add(self, ctx):"""增@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""body = ctx.bodyunique = self.config.get("unique")obj = Noneif unique:qy = {}for i in range(len(unique)):key = unique[i]qy[key] = body.get(key)obj = self.service.Get_obj(qy)if not obj:# 添加数据前error = self.Add_before(ctx)if error["code"]:return {"error": error}error = self.Events("add_before", ctx, None)if error["code"]:return {"error": error}# 添加数据result = self.service.Add(body, self.config)# 添加数据发生错误if self.service.error:return {"error": self.service.error}# 添加数据成功后res = self.Add_after(ctx, result)if res:result = resres = self.Events("add_after", ctx, result)if res:result = resreturn {"result": result}else:return {"error": {"code": 10000, "message": "已存在"}}# 删def Del(self, ctx):"""删@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""if len(ctx.query) == 0:errorMsg = {"code": 30000, "message": "删除条件不能为空！"}return errorMsgresult = self.service.Del(ctx.query, self.config)if self.service.error:return {"error": self.service.error}return {"result": result}# 改def Set(self, ctx):"""改@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""# 修改数据前error = self.Set_before(ctx)if error["code"]:return {"error": error}error = self.Events("set_before", ctx, None)if error["code"]:return {"error": error}query = ctx.queryif 'page' in query.keys():del ctx.query['page']if 'size' in query.keys():del ctx.query['size']if 'orderby' in query.keys():del ctx.query['orderby']# 修改数据result = self.service.Set(ctx.query, ctx.body, self.config)# 修改数据发生错误if self.service.error:return {"error": self.service.error}# 修改数据成功后res = self.Set_after(ctx, result)if res:result = resres = self.Events("set_after", ctx, result)if res:result = resreturn {"result": result}# 查多条def Get_list(self, ctx):"""查多条@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldif "page" in query:config_plus["page"] = query.pop("page")if "size" in query:config_plus["size"] = query.pop("size")if "orderby" in query:config_plus["orderby"] = query.pop("orderby")if "like" in query:config_plus["like"] = query.pop("like")if "groupby" in query:config_plus["groupby"] = query.pop("groupby")count = self.service.Count(query)lst = []if self.service.error:return {"error": self.service.error}elif count:lst = self.service.Get_list(query,obj_update(self.config, config_plus))if self.service.error:return {"error": self.service.error}self.interact_list(ctx, lst)return {"result": {"list": lst, "count": count}}# 查一条def Get_obj(self, ctx):"""查一条@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldobj = self.service.Get_obj(query, obj_update(self.config, config_plus))if self.service.error:return {"error": self.service.error}if obj:self.interact_obj(ctx, obj)return {"result": {"obj": obj}}# 饼图统计def List_group(self, ctx):"""饼图统计@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""query = dict(ctx.query)config_plus = {}if "groupby" in query:groupby_t = query.pop("groupby")config_plus["groupby"] = groupby_telse:err = {"error": 30000, "message": "groupby的值不能为空！"}return errlt = self.service.Count_group(query, obj_update(self.config, config_plus))for o in lt:o[1] = o[groupby_t]o[0] = o["count"]if self.service.error:return {"error": self.service.error}return {"result": {"list": lt}}# 柱状图统计def Bar_group(self, ctx):"""柱状图统计@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldelse:err = {"error": 30000, "message": "field的值不能为空！"}return errif "groupby" in query:groupby_t = query.pop("groupby")config_plus["groupby"] = groupby_telse:err = {"error": 30000, "message": "groupby的值不能为空！"}return errlt = self.service.Bar_group(query, obj_update(self.config, config_plus))for k, v in enumerate(lt):new = list(v.values())lt[k] = newif self.service.error:return {"error": self.service.error}return {"result": {"list": lt}}# 总数def Count(self, ctx):"""总数@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""result = self.service.Count(ctx.query, self.config)if self.service.error:return {"error": self.service.error}return {"result": result}# 分组总计条数def Count_group(self, ctx):"""分组总计条数@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""query = dict(ctx.query)config_plus = {}if "groupby" in query:groupby_t = query.pop("groupby")config_plus["groupby"] = groupby_telse:err = {"error": 30000, "message": "groupby的值不能为空！"}return errlt = self.service.Count_group(query,obj_update(self.config, config_plus))if self.service.error:return {"error": self.service.error}return {"result": {"list": lt}}# 合计def Sum(self, ctx):"""合计@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldelse:err = {"error": 30000, "message": "field的值不能为空！"}return errresult = self.service.Sum(query, obj_update(self.config, config_plus))if self.service.error:return {"error": self.service.error}return {"result": result}# 分组求和def Sum_group(self, ctx):"""分组求和@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldelse:err = {"error": 30000, "message": "field的值不能为空！"}return errif "groupby" in query:groupby_t = query.pop("groupby")config_plus["groupby"] = groupby_telse:err = {"error": 30000, "message": "groupby的值不能为空！"}return errlt = self.service.Sum_group(query,obj_update(self.config, config_plus))if self.service.error:return {"error": self.service.error}return {"result": {"list": lt}}# 求平均数def Avg(self, ctx):"""求平均数@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldelse:err = {"error": 30000, "message": "field的值不能为空！"}return errresult = self.service.Avg(query, obj_update(self.config, config_plus))if self.service.error:return {"error": self.service.error}return {"result": result}# 分组平均数def Avg_group(self, ctx):"""分组平均数@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""query = dict(ctx.query)config_plus = {}if "field" in query:field = query.pop("field")config_plus["field"] = fieldelse:err = {"error": 30000, "message": "field的值不能为空！"}return errif "groupby" in query:groupby_t = query.pop("groupby")config_plus["groupby"] = groupby_telse:err = {"error": 30000, "message": "groupby的值不能为空！"}return errlt = self.service.Avg_group(query,obj_update(self.config, config_plus))if self.service.error:return {"error": self.service.error}return {"result": {"list": lt}}# 导入数据def Import_db(self, ctx):"""导入数据@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""body = {"error": {"code": 10000, "message": "未定义表名！"}}return body# 导出数据def Export_db(self, ctx):"""导出数据@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""message = {"id": 1,"jsonrpc": "2.0",}query = ctx.query# 获取表名table = self.config["services"]path = ""if "path" in query:path = query.pop("path")if "name" in query:name = query.pop("name")# 通过服务获得数据service = service_select(table)lst = service.Export_db(query)# 1.创建文件对象f = open(str(path) + str(name) + ".csv","w",newline="",encoding="utf-8")# 2.基于文件对象构建 csv写入对象csv_writer = csv.writer(f)for row in lst:csv_writer.writerow(row)return message# 上传def Upload(self, ctx):"""上传@param {Object} ctx http请求上下文@return {Object} 返回json-rpc格式结果"""file_obj = ctx.request.files.get("file", None)if file_obj is None:error = {"code": 10000, "message": "上传的文件(file)不能为空"}return errortry:file_obj = ctx.request.files.get("file", None)u = "static/upload/" + file_obj.filenamefile_obj.save(os.path.join(u))# fe = os.getcwd() + u# with open(fe, "wb") as f:#     for line in file_obj.chunks():#         f.write(line)# f.close()except Exception as e:print("上传失败：", e.args)return {'error':e.args}else:return {"result": {"url": "/"+u}}# 鉴权def Auth(self, ctx):if len(dict_auth.keys()) == 0:service = service_select("auth")lst = service.Get_list({}, {"page": 0})for o in lst:if "option" in o:o["option"] = ast.literal_eval(o["option"])else:o["option"] = {}path = o["path"]if not dict_auth[path]:dict_auth[path] = {}dict_auth[path][o["user_group"]] = oreturn dict_auth# 添加前def Add_before(self, ctx):# print("添加", ctx)return {"code": 0}# 添加后def Add_after(self, ctx, result):# print("结果", ctx)return result# 修改前def Set_before(self, ctx):# print("修改前", ctx)return {"code": 0}# 分类推荐def Get_hits_list(self, ctx):return {"code": 0}# 商家查询def Get_business_order_list(self, ctx):return {"code": 0}# 修改前def Set_after(self, ctx, result):return result# 事件def Events(self, event_name, param, paramB):if event_name == "add_before":return {"code": 0}elif event_name == "del_before":return {"code": 0}elif event_name == "set_before":return {"code": 0}elif event_name == "get_obj_before":return {"code": 0}elif event_name == "get_list_before":return {"code": 0}elif event_name == "add_after":return paramBelif event_name == "del_after":return paramBelif event_name == "set_after":return paramBelif event_name == "get_obj_after":return paramBelif event_name == "get_list_after":return paramBelse:return paramB