谨以此博客作为记录
小编这里用的版本是：
<hadoop.version>2.7.7</hadoop.version>
<spark.version>2.4.5</spark.version>
<scala.version>2.12.10</scala.version>
如果没用到Hadoop可以忽略
步骤
- 准备数据，知道每个文件什么内容，每一列什么意思
我这里对应文件夹下，是按日期作为文件名存储的txt文件数据
- 数据可放在本地，也可放在HDFS上，数据在本地代码完全是可以跑通的；但是小编想模拟数据存在集群（HDFS）上的情况，条件有限，DataNode总是dead，搞得小编也很sad（押上了hhh）
- 编写代码
- ①首先创建SparkContext和SparkSession对象；
- ②引入implicit隐式转换包，方便后面将RDD转换为DataFrame；
- ③创建数据库连接，方便后面将结果写入数据库；
- ④从文件名中截取日期并写入文件，后面分析数据需要；
- ⑤利用RDD的算子将数据进行拆分、切割、整理，并转换为DataFrame；
- ⑥创建临时表，便于用SQL语句对表进行查询；
- ⑦将查询结果写入数据库；
代码
import org.apache.spark.sql.{SparkSession}
import org.apache.spark.{SparkConf, SparkContext}object readData {def main(args: Array[String]): Unit = {//SparkConf是用来配置Spark应用程序的参数的类。在这个代码中，//setAppName("ReadData")设置了应用程序的名称为"ReadData"，//而setMaster("local[*]")指定了Spark应用程序的运行模式为本地模式，使用所有可用的CPU核心val conf = new SparkConf().setAppName("ReadData").setMaster("local[*]")//初始化一个SparkContext对象。SparkContext是与Spark集群通信的主要入口点val sc = new SparkContext(conf)//创建了一个SparkSession对象，它是Spark 2.0引入的API，提供了一种统一的入口点来访问Spark功能val spark = SparkSession.builder().getOrCreate()import spark.implicits._//创建数据库连接var prop=new java.util.Properties()prop.put("user","root") //MySQL用户名prop.put("password","123456") //MySQL密码prop.put("driver","com.mysql.jdbc.Driver") //MySQL驱动val url="jdbc:mysql://localhost:3306/login" MySQL URL,端口号后面是数据库名（login）//如果数据在HDFS集群上，就可以将路径替换为HDFS集群上的路径//val hdfsPath = "hdfs://IP:9000/data/0007/CharacterLogin";//路径val path="\\2015-6-12\\0001\\0007\\CharacterLogin"//wholeTextFiles: 是SparkContext对象的一个方法，用于读取整个文本文件//返回的RDD: 其中键是文件路径，值是文件内容。val rdd1 = sc.wholeTextFiles(path)
// 打印文件路径进行调试
// rdd1.foreach { t =>
// val filePath = t._1
// println("File Path: " + filePath)
// }//从文件名中截取日期并写入文件//rdd2返回的是alllines拼接之后的数据val rdd2 = rdd1.flatMap(t=>{val filePath = t._1val alllines = t._2//使用了 substring 方法来截取文件路径中从最后一个 / 符号（包括）之后到 .txt 扩展名（不包括）之前的部分val dateStr=filePath.substring(filePath.lastIndexOf("/")+1, filePath.lastIndexOf(".txt"))alllines.split("\r\n").map(line=>dateStr+"|"+line)})//对数据进行拆分，提取相关信息// 这里主要统计人均登陆次数（DEC）==总的登录次数除以用户数// LoginStatus==0表示登录val res = rdd2.map(line => line.split('|')).map(arr => (arr(6), arr(11),arr(0))).toDF("userID", "LoginStatus","loginDate").filter("LoginStatus==0")//临时表创建res.createTempView("LoginCount")//查询val res1=spark.sql("select loginDate,count(LoginStatus)/count(distinct userID) as loginNum from LoginCount where LoginStatus=0 group by loginDate").sort("loginDate")//写入mysql数据库res1.write.mode("append").jdbc(url,"logincount",prop)}
}
