pil3aoekg
/
javac
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

1.环境搭建完成11

Browse Source 
2.相关的工具类编写完成11
3.配置文件管理类编写完成11
zxr
劝酒千百 9 months ago
parent b357bb9ef6
commit 0af0f9bd61
2 changed files with 112 additions and 6 deletions
Show Stats Download Patch File Download Diff File

16
.idea/workspace.xml
Unescape View File

@ -4,7 +4,10 @@
<option name="autoReloadType" value="ALL" />
</component>
<component name="ChangeListManager">
<list default="true" id="a7505764-040b-48e2-b2fc-8c5b579e595f" name="Default" comment="1.环境搭建完成&#10;2.相关的工具类编写完成11&#10;3.配置文件管理类编写完成11" />
<list default="true" id="a7505764-040b-48e2-b2fc-8c5b579e595f" name="Default" comment="1.环境搭建完成&#10;2.相关的工具类编写完成11&#10;3.配置文件管理类编写完成11">
<change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
<change beforePath="$PROJECT_DIR$/src/main/java/cn/edu/hust/session/UserVisitAnalyze.java" beforeDir="false" afterPath="$PROJECT_DIR$/src/main/java/cn/edu/hust/session/UserVisitAnalyze.java" afterDir="false" />
</list>
<option name="SHOW_DIALOG" value="false" />
<option name="HIGHLIGHT_CONFLICTS" value="true" />
<option name="HIGHLIGHT_NON_ACTIVE_CHANGELIST" value="false" />
@ -474,6 +477,7 @@
<workItem from="1734349398416" duration="1738000" />
<workItem from="1734352550638" duration="1153000" />
<workItem from="1734354068325" duration="860000" />
<workItem from="1734354940766" duration="2406000" />
</task>
<task id="LOCAL-00001" summary="1.环境搭建完成&#10;2.相关的工具类编写完成&#10;3.配置文件管理类编写完成">
<created>1529592741848</created>
@ -537,7 +541,15 @@
<option name="project" value="LOCAL" />
<updated>1734354900465</updated>
</task>
<option name="localTasksCounter" value="9" />
<task id="LOCAL-00009" summary="1.环境搭建完成&#10;2.相关的工具类编写完成11&#10;3.配置文件管理类编写完成11">
<option name="closed" value="true" />
<created>1734354969705</created>
<option name="number" value="00009" />
<option name="presentableId" value="LOCAL-00009" />
<option name="project" value="LOCAL" />
<updated>1734354969705</updated>
</task>
<option name="localTasksCounter" value="10" />
<servers />
</component>
<component name="TestHistory">

102
src/main/java/cn/edu/hust/session/UserVisitAnalyze.java
Unescape View File

@ -1,25 +1,94 @@
//这行代码定义了当前类所在的包名。包名用于组织和管理类。
package cn.edu.hust.session;
//导入 cn.edu.hust.conf 包下的 ConfigurationManager 类。这个类可能用于管理配置信息
import cn.edu.hust.conf.ConfigurationManager;
//导入 cn.edu.hust.constant 包下的 Constants 类。这个类可能包含一些常量定义。
import cn.edu.hust.constant.Constants;
//导入 cn.edu.hust.dao 包下的 TaskDao 类。这个类可能用于与任务相关的数据访问操作。
import cn.edu.hust.dao.TaskDao;
//导入 cn.edu.hust.dao.factory 包下的 DaoFactory 类。这个类可能用于创建 DAO 对象。
import cn.edu.hust.dao.factory.DaoFactory;
//导入 cn.edu.hust.domain 包下的所有类。这个包可能包含一些数据模型类。
import cn.edu.hust.domain.*;
//导入 cn.edu.hust.mockData 包下的 MockData 类。这个类可能用于生成模拟数据
import cn.edu.hust.mockData.MockData;
//导入 cn.edu.hust.util 包下的所有类。这个包可能包含一些通用工具类。
import cn.edu.hust.util.*;
//导入 com.alibaba.fastjson.JSONObject 类。FastJSON 是一个用于处理 JSON 数据的库。
import com.alibaba.fastjson.JSONObject;
//导入 org.apache.spark.Accumulator 类。这个类用于在 Spark 作业中共享和累加数据。
import org.apache.spark.Accumulator;
//导入 org.apache.spark.SparkConf 类。这个类用于配置 Spark 应用程序。
import org.apache.spark.SparkConf;
//导入 org.apache.spark.SparkContext 类。这个类是 Spark 应用程序的主入口点。
import org.apache.spark.SparkContext;
//导入 org.apache.spark.api.java.JavaPairRDD 类。
// 这个类是用于处理键值对的 RDD（弹性分布式数据集）
import org.apache.spark.api.java.JavaPairRDD;
//导入 org.apache.spark.api.java.JavaRDD 类。这个类是用于处理普通 RDD 的 Java API。
import org.apache.spark.api.java.JavaRDD;
//导入了 JavaSparkContext 类。
// JavaSparkContext 是 org.apache.spark.api.java 包下的一个类，提供了与 Spark 集群交互的 Java API。
import org.apache.spark.api.java.JavaSparkContext;
//导入 org.apache.spark.api.java.function 包下的所有函数类。
// 这些类提供了常见的操作函数，如 FlatMapFunction, MapFunction, FilterFunction 等。
import org.apache.spark.api.java.function.*;
//导入 org.apache.spark.sql.DataFrame 类。这个类是用于操作结构化数据的。
import org.apache.spark.sql.DataFrame;
//导入 org.apache.spark.sql.Row 类。这个类是用于表示一行数据的。
import org.apache.spark.sql.Row;
//导入 org.apache.spark.sql.SQLContext 类。
// 这个类用于创建 Spark SQL 的上下文环境
import org.apache.spark.sql.SQLContext;
//导入 org.apache.spark.sql.hive.HiveContext 类。
//这个类用于与 Hive 兼容的 Spark SQL 上下文环境
import org.apache.spark.sql.hive.HiveContext;
//导入 org.apache.spark.storage.StorageLevel 类。
// 这个类定义了 Spark 中数据存储的级别。
import org.apache.spark.storage.StorageLevel;
//导入 scala.Tuple2 类。这个类用于表示一个二元组（一个包含两个元素的元组）。
import scala.Tuple2;
import java.util.*;
@ -36,50 +105,75 @@ import java.util.*;
public class UserVisitAnalyze {
public static void main(String[] args)
{
//初始化 args 参数数组，用于传递给 main 方法。
// 这里暂时设置为 {"1"}，实际使用时可以根据需要调整。
args=new String[]{"1"};
/**
* 构建spark上下文
*/
//创建 SparkConf 对象并设置应用程序名称和运行模式（本地模式，使用3个核心）。
//使用 SparkConf 创建 JavaSparkContext 实例，初始化Spark上下文。
SparkConf conf=new SparkConf().setAppName(Constants.APP_NAME_SESSION).setMaster("local[3]");
JavaSparkContext context=new JavaSparkContext(conf);
//通过 JavaSparkContext 获取 SQLContext，用于执行SQL查询。
SQLContext sc=getSQLContext(context.sc());
//生成模拟数据
//生成模拟数据,调用 mock 方法生成模拟数据到Spark环境中。
mock(context,sc);
//拿到相应的Dao组建
//通过 DaoFactory 获取 TaskDao 实例，用于数据库操作。
TaskDao dao= DaoFactory.getTaskDao();
//从外部传入的参数获取任务的id
//通过 args 参数获取任务ID。
Long taskId=ParamUtils.getTaskIdFromArgs(args);
//从数据库中查询出相应的task
//通过任务ID从数据库中查询任务信息，并将任务参数解析为 JSONObject
Task task=dao.findTaskById(taskId);
JSONObject jsonObject=JSONObject.parseObject(task.getTaskParam());
//获取指定范围内的Sesssion
//调用 getActionRDD 方法获取包含指定范围内的Session的RDD。
JavaRDD<Row> sessionRangeDate=getActionRDD(sc,jsonObject);
//这里增加一个新的方法，主要是映射
//将 sessionRangeDate 转换为包含键值对的 PairRDD。
JavaPairRDD<String,Row> sessionInfoPairRDD=getSessonInfoPairRDD(sessionRangeDate);
//重复用到的RDD进行持久化
//将 sessionInfoPairRDD 持久化到磁盘，提高后续操作的性能
sessionInfoPairRDD.persist(StorageLevel.DISK_ONLY());
//上面的两个RDD是
//按照Sesson进行聚合
//调用 aggregateBySessionId 方法对Session信息进行聚合。
JavaPairRDD<String,String> sesssionAggregateInfoRDD=aggregateBySessionId(sc,sessionInfoPairRDD);
//通过条件对RDD进行筛选
// 重构，同时统计
//使用Accumulator进行统计：
//创建一个 Accumulator 来统计聚合结果。
Accumulator<String> sessionAggrStatAccumulator=context.accumulator("",new SessionAggrStatAccumulator());
//过滤和统计Session信息：
//在进行accumulator之前，需要aciton动作，不然会为空
//调用 filterSessionAndAggrStat 方法过滤并统计Session信息。
JavaPairRDD<String,String> filteredSessionRDD=filterSessionAndAggrStat(sesssionAggregateInfoRDD,jsonObject,sessionAggrStatAccumulator);
//重复用到的RDD进行持久化
//将过滤后的RDD持久化到磁盘。
filteredSessionRDD.persist(StorageLevel.DISK_ONLY());
//获取符合过滤条件的全信息公共RDD
//调用 getFilterFullInfoRDD 方法获取包含完整信息的公共RDD。
JavaPairRDD<String, Row> commonFullClickInfoRDD=getFilterFullInfoRDD(filteredSessionRDD,sessionInfoPairRDD);
//重复用到的RDD进行持久化
//将公共RDD持久化到磁盘。
commonFullClickInfoRDD.persist(StorageLevel.DISK_ONLY());
//session聚合统计，统计出访问时长和访问步长的各个区间所占的比例
/**
* 重构实现的思路：
* 1。不要去生成任何的新RDD

Write Preview
Loading…
Cancel
Save