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

天空闲话 · 2025-3-6 23:24:01

作者简介：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_select
mysqlPool = 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.query
m["body"] = ctx.body
m["auth"] = ctx.auth
return 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"] = field
if "page" in query:
page = query.pop("page")
config_plus["page"] = page
if "size" in query:
size = query.pop("size")
config_plus["size"] = size
result_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"] = field
if "page" in query:
page = query.pop("page")
config_plus["page"] = page
if "size" in query:
size = query.pop("size")
config_plus["size"] = size
result_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"] = field
if "page" in query:
page = query.pop("page")
config_plus["page"] = page
if "size" in query:
size = query.pop("size")
config_plus["size"] = size
result_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"] = field
obj_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"] = field
obj_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.body
unique = self.config.get("unique")
obj = None
if 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 = res
res = self.Events("add_after", ctx, result)
if res:
result = res
return {"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 errorMsg
result = 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.query
if '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 = res
res = self.Events("set_after", ctx, result)
if res:
result = res
return {"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"] = field
if "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"] = field
obj = 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_t
else:
err = {"error": 30000, "message": "groupby的值不能为空！"}
return err
lt = 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"] = field
else:
err = {"error": 30000, "message": "field的值不能为空！"}
return err
if "groupby" in query:
groupby_t = query.pop("groupby")
config_plus["groupby"] = groupby_t
else:
err = {"error": 30000, "message": "groupby的值不能为空！"}
return err
lt = self.service.Bar_group(query, obj_update(self.config, config_plus))
for k, v in enumerate(lt):
new = list(v.values())
lt[k] = new
if 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_t
else:
err = {"error": 30000, "message": "groupby的值不能为空！"}
return err
lt = 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"] = field
else:
err = {"error": 30000, "message": "field的值不能为空！"}
return err
result = 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"] = field
else:
err = {"error": 30000, "message": "field的值不能为空！"}
return err
if "groupby" in query:
groupby_t = query.pop("groupby")
config_plus["groupby"] = groupby_t
else:
err = {"error": 30000, "message": "groupby的值不能为空！"}
return err
lt = 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"] = field
else:
err = {"error": 30000, "message": "field的值不能为空！"}
return err
result = 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"] = field
else:
err = {"error": 30000, "message": "field的值不能为空！"}
return err
if "groupby" in query:
groupby_t = query.pop("groupby")
config_plus["groupby"] = groupby_t
else:
err = {"error": 30000, "message": "groupby的值不能为空！"}
return err
lt = 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 error
try:
file_obj = ctx.request.files.get("file", None)
u = "static/upload/" + file_obj.filename
file_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"]] = o
return 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 paramB
elif event_name == "del_after":
return paramB
elif event_name == "set_after":
return paramB
elif event_name == "get_obj_after":
return paramB
elif event_name == "get_list_after":
return paramB
else:
return paramB

复制代码

五，相关作品展示

基于Java开辟、Python开辟、PHP开辟、C#开辟等相关语言开辟的实战项目
基于Nodejs、Vue等前端技术开辟的前端实战项目
基于微信小程序和安卓APP应用开辟的相关作品
基于51单片机等嵌入式物联网开辟应用
基于各类算法实现的AI智能应用
基于大数据实现的各类数据管理和推荐系统