计算机毕业设计hadoop+spark天气预测 天气可视化 天气大数据 空气质量检测 空气质量分析 气象大数据 气象分析 大数据毕业设计 大数据毕设_基于hadoop天气数据分析系统报告论文

毕业设计开题报告

任务书

一 系统功能

本天气数据分析系统，主要有数据采集，数据存储，数据预处理，数据分析，数据可视化，预测模型等类型。

具体实现目标如下：

(1).数据采集模块：这个模块负责从天气网爬取天气数据，并将其传输到Hadoop或Spark集群进行处理。

(2).数据存储模块：这个模块负责将处理后的数据以表结构形式存储数据库中，以便后续查询和分析。

(3).数据预处理模块：这个模块负责对采集到的原始数据进行清洗、转换和标准化等操作，以便后续的分析和处理。

(4).数据分析模块：这个模块使用Hadoop和Spark的大数据处理能力，对经过预处理的数据进行分析，以提取天气模式、趋势和预测等有用信息。

(5).数据可视化模块：这个模块将分析结果以图表、图形等形式展示给用户，以便他们能够直观地了解天气状况和趋势。

(6).预测模型模块：这个模块利用线性回归预测，对天气数据进行预测和分析，提供未来天气状况的预测结果。

二、实现方式

该项目采用Hadoop,Spark,Zookeeper, kafka,Hbase等技术搭建了一个基于Spark的天气分析预测系统，具体的实现模块分为以下几部分：

1. 使Python和PyCharm进行数据采集，从天气网爬取天气数据。
2. 数据库使用MySQL，来存取项目数据，以表结构的形式存在。
3. 数据分析采用Sparksql，分析不同影响因素。
4. 可视化采用Matplotlib在Tableau上绘制出柱状图折线图和饼图。
5. 预测模型采用线性回归模型进行训练。使用Python中的机器学习库将预处理后的        数据集作为输入，并使用标签数据（例如未来一段时间内的天气状况）作为输出。

核心算法代码分享如下：

https://dblab.xmu.edu.cn/post/8116/
 
## 启动hadoop
cd /data/hadoop/sbin
start-all.sh
 
 
 
## 启动hive 
 
cd /data/hive
 
nohup hive --service metastore &
 
nohup hive --service hiveserver2 &
 
 
head -5 user_log.csv
 
 
sed -i '1d' user_log.csv
 
 
predeal.sh
 
chmod +x ./predeal.sh
./predeal.sh ./user_log.csv ./small_user_log.csv
 
 
 
hadoop dfs -mkdir -p /dbtaobao/dataset/user_log
hadoop dfs -put dfs -put /data/ebiz2024/small_user_log.csv /dbtaobao/dataset/user_log
hadoop dfs -cat /dbtaobao/dataset/user_log/small_user_log.csv | head -10
 
drop database if exists dbtaobao;
create database if not exists dbtaobao;
use dbtaobao;
 
 
CREATE EXTERNAL TABLE dbtaobao.user_log(user_id INT,item_id INT,cat_id INT,merchant_id INT,brand_id INT,month STRING,day STRING,action INT,age_range INT,gender INT,province STRING) COMMENT 'Welcome to xmu dblab,Now create dbtaobao.user_log!' ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' STORED AS TEXTFILE LOCATION '/dbtaobao/dataset/user_log';
 
 
 
 
select * from user_log limit 10;
 
use dbtaobao;
 show tables;
 
  show create table user_log;
  desc user_log;
  
  select month,day,cat_id from user_log limit 20;
  select brand_id from user_log limit 10; -- 查看日志前10个交易日志的商品品牌
  
  select ul.at, ul.ci  from (select action as at, cat_id as ci from user_log) as ul limit 20;
   select count(*) from user_log;
   select count(distinct user_id) from user_log; -- 在函数内部加上distinct，查出user_id不重复的数据有多少条
   
   
   select count(*) from (select user_id,item_id,cat_id,merchant_id,brand_id,month,day,action from user_log group by user_id,item_id,cat_id,merchant_id,brand_id,month,day,action having count(*)=1)a;
   
   
   select count(distinct user_id) from user_log where action='2';
   
   。。。。。。。。。。。。。。。。。。。
   
   
   
   hive> create table scan(brand_id INT,scan INT) COMMENT 'This is the search of bigdatataobao' ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' STORED AS TEXTFILE; -- 创建新的数据表进行存储
hive> insert overwrite table scan select brand_id,count(action) from user_log where action='2' group by brand_id; --导入数据
hive> select * from scan; -- 显示结果
   
  select * from scan limit 10; 
  
  
  hive> create table dbtaobao.inner_user_log(user_id INT,item_id INT,cat_id INT,merchant_id INT,brand_id INT,month STRING,day STRING,action INT,age_range INT,gender INT,province STRING) COMMENT 'Welcome to XMU dblab! Now create inner table inner_user_log ' ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' STORED AS TEXTFILE;
 
 hive> INSERT OVERWRITE TABLE dbtaobao.inner_user_log select * from dbtaobao.user_log;
 
 select * from inner_user_log limit 10;
 
 
 
 CREATE TABLE `dbtaobao`.`user_log` (`user_id` varchar(20),`item_id` varchar(20),`cat_id` varchar(20),`merchant_id` varchar(20),`brand_id` varchar(20), `month` varchar(6),`day` varchar(6),`action` varchar(6),`age_range` varchar(6),`gender` varchar(6),`province` varchar(10)) ENGINE=InnoDB DEFAULT CHARSET=utf8;
 
 
 
 sqoop export --connect jdbc:mysql://localhost:3306/dbtaobao --username root --password 123456 --table user_log --export-dir '/user/hive/warehouse/dbtaobao.db/inner_user_log' --fields-terminated-by ',';
 
 
 
 
hadoop fs -put /data/ebiz2024/train_after.csv /dbtaobao/dataset
hadoop fs -put /data/ebiz2024/test_after.csv /dbtaobao/dataset
 
 
 
cd /data/jars/spark
/data/spark/bin/spark-shell --jars /data/jars/spark/mysql-connector-java-5.1.47.jar --driver-class-path /data/jars/spark/mysql-connector-java-5.1.47.jar
 
 
 
 
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.linalg.{Vectors,Vector}
import org.apache.spark.mllib.classification.{SVMModel, SVMWithSGD}
import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics
import java.util.Properties
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
val train_data = sc.textFile("/data/ebiz2024/train_after.csv")
val test_data = sc.textFile("/data/ebiz2024/test_after.csv")
val train= train_data.map{line =>
  val parts = line.split(',')
  LabeledPoint(parts(4).toDouble,Vectors.dense(parts(1).toDouble,parts
(2).toDouble,parts(3).toDouble))
}
val test = test_data.map{line =>
  val parts = line.split(',')
  LabeledPoint(parts(4).toDouble,Vectors.dense(parts(1).toDouble,parts(2).toDouble,parts(3).toDouble))
}
val numIterations = 1000
val model = SVMWithSGD.train(train, numIterations)
model.clearThreshold()
val scoreAndLabels = test.map{point =>
  val score = model.predict(point.features)
  score+" "+point.label
}
scoreAndLabels.foreach(println)
model.setThreshold(0.0)
scoreAndLabels.foreach(println)
model.clearThreshold()
val scoreAndLabels = test.map{point =>
  val score = model.predict(point.features)
  score+" "+point.label
}
 
 
//设置回头客数据
val rebuyRDD = scoreAndLabels.map(_.split(" "))
/下面要设置模式信息
val schema = StructType(List(StructField("score", StringType, true),StructField("label", StringType, true)))
//下面创建Row对象，每个Row对象都是rowRDD中的一行
val rowRDD = rebuyRDD.map(p => Row(p(0).trim, p(1).trim))
//建立起Row对象和模式之间的对应关系，也就是把数据和模式对应起来
val rebuyDF = spark.createDataFrame(rowRDD, schema)
//下面创建一个prop变量用来保存JDBC连接参数
val prop = new Properties()
prop.put("user", "root") //表示用户名是root
prop.put("password", "123456") //表示密码是hadoop
prop.put("driver","com.mysql.jdbc.Driver") //表示驱动程序是com.mysql.jdbc.Driver
//下面就可以连接数据库，采用append模式，表示追加记录到数据库dbtaobao的rebuy表中
rebuyDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/dbtaobao", "dbtaobao.rebuy", prop) 
 
 
 
 
val rebuyRDD = scoreAndLabels.map(_.split(" "))
 
val schema = StructType(List(StructField("score", StringType, true),StructField("label", StringType, true)))
 
val rowRDD = rebuyRDD.map(p => Row(p(0).trim, p(1).trim))
 
val rebuyDF = spark.createDataFrame(rowRDD, schema)
 
val prop = new Properties()
prop.put("user", "root")
prop.put("password", "123456") 
prop.put("driver","com.mysql.jdbc.Driver")
 
rebuyDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/dbtaobao", "dbtaobao.rebuy", prop)