package com.aspirecn.jiaoke.data.extract import java.nio.file.{Files, Paths} import java.sql.Date import org.apache.log4j.{Level, Logger} import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession} import org.apache.spark.storage.StorageLevel import scala.collection.immutable /** * @auther: hwj * @date: 2021/4/13 10:39 * @description: */ object DataExtractByReadFile { case class TrcExetcpu(vehicleid:String, exTime:Date, enTime:Date,feemileage :String, shortfeemileage:String,fee:String, enweight:String, extolllaneid:String, axlecount:String, vehicletype:String, day:String ) case class GBUPLoad(vehicleid:String, transtime:String, day:String) def main(args: Array[String]): Unit = { val log: Logger = Logger.getLogger("DataExtractByReadFile") log.setLevel(Level.INFO) if (args.size != 6){ log.error( """ |请正确传入程序参数：1.门架数据读取目录 | 2.ETC数据读取目录 | 3.现金出口数据读取目录 | 4.结果文件的输出路径（文件输出目录在执行不可存在，程序会自动进行创建，否则会报错） | 5.checkpoint目录 | 6.结果文件输出个数，请根据数据量合理输入该参数，输出的文件越少，执行时间越长，输入0则为默认文件个数 |""".stripMargin) System.exit(0) } // 门架数据读取目录 val menjiaPath: String = args(0) // etc数据读取目录 val etcPath: String = args(1) // 现金出口数据读取目录 val qitachukouPath: String = args(2) // 结果文件输出目录 val out_path: String = args(3) // checkpoint目录 val check_path: String = args(4) // 结果文件输出个数 val result_partitions: Int = Integer.valueOf(args(5)) // 当数据读取目录不存在时，结束程序 if((!exists(menjiaPath))||(!exists(etcPath))||(!exists(qitachukouPath))){ log.error("请输入正确的文件目录:" + "高架数据路径:"+menjiaPath + "etc数据路径:"+etcPath + "现金出口数据路径:"+qitachukouPath) System.exit(0) } // 当结果数据输出目录已存在时，结束程序 if(exists(out_path)){ log.error("数据输出目录已存在"+out_path) System.exit(0) } // 初始化SparkSession val spark: SparkSession = formatSparkSession //设置checkpoint spark.sparkContext.setCheckpointDir(check_path) // spark.sparkContext.setCheckpointDir("file:///data/checkpoint") log.info("开始抽取数据") //todo：开始处理数据 import org.apache.spark.sql.functions._ // todo: 引入隐式转换 import spark.implicits._ // 高架数据 log.info("开始抽取门架数据，抽取路径为："+menjiaPath) val gbuLoadDataset: Dataset[GBUPLoad] = spark.read.option("header", true).csv(menjiaPath).toDF("vehicleid", "transtime", "day").as[GBUPLoad].coalesce(36) log.info("抽取门架数据成功") //etc数据 log.info("开始抽取etc出口数据，抽取路径为："+etcPath) val trcExetcpuDataset = spark.read.option("header", true).csv(etcPath).toDF("vehicleid","extime","entime","feemileage","shortfeemileage","fee","enweight","extolllaneid","axlecount","vehicletype","day").as[TrcExetcpu].coalesce(36) log.info("抽取etc出口数据成功") // 其他出口数据 log.info("抽取开始抽取现金出口数据，抽取路径为："+qitachukouPath) val trcOtuDataset = spark.read.option("header", true).csv(qitachukouPath).toDF("vehicleid","extime","entime","feemileage","shortfeemileage","fee","enweight","extolllaneid","axlecount","vehicletype","day").as[TrcExetcpu].coalesce(36) log.info("现金出口数据成功") //合并etc出口和现金出口的数据 log.info("开始合并etc出口和现金出口数据") val trcDataSet: Dataset[TrcExetcpu] = trcExetcpuDataset.unionByName(trcOtuDataset) log.info("合并etc出口和现金出口数据成功") // 创建视图 对出口数据进行处理 trcDataSet.createOrReplaceTempView("trcDataSet") log.info("创建视图，开始统计出口相关数据") val trcDataSet_p: DataFrame = spark.sql( """ |select |vehicleid, |exTime, |entime, |(case when entime is not null then unix_timestamp(regexp_replace(extime,'T',' '))-unix_timestamp(regexp_replace(entime,'T',' ')) else 0 end ) as travel_time, |(case when feemileage>0 and feemileage is not null then feemileage else shortfeemileage end) as feemileage, | cast(fee as double) as fee, |(case when enweight >0 then ( case when feemileage>0 and feemileage is not null then feemileage else shortfeemileage end) else 0 end ) as weight_mileage, |substr(exTollLaneId,6,2) as travel_provinces, |axlecount, |vehicletype, |day |from |trcDataSet | |""".stripMargin) log.info("出口相关数据统计完成") // 取出出口表的 出口交易时间和入口交易时间，与门架表进行合并 log.info("开始抽取出口数据中的出口交易时间") val ex_transtime: Dataset[GBUPLoad] = trcDataSet_p.select($"day", $"vehicleid".as("vehicleid"), $"exTime".as("transtime")).as[GBUPLoad] log.info("抽取出口数据中的出口交易时间成功") log.info("开始抽取出口数据中的入口交易时间") val en_transtime: Dataset[GBUPLoad] = trcDataSet_p.select($"day", $"vehicleid".as("vehicleid"),$"enTime".as("transtime")).as[GBUPLoad] log.info("抽取出口数据中的入口交易时间成功") log.info("开始合并门架数据时间与 入口交易时间、出口交易时间") val exetcpuDataset: Dataset[GBUPLoad] = gbuLoadDataset.unionByName(ex_transtime).unionByName(en_transtime) log.info("合并交易时间成功") trcDataSet_p.persist(StorageLevel.MEMORY_AND_DISK) trcDataSet_p.checkpoint() trcDataSet_p.unpersist() // 创建视图 对用户出行数据进行分段统计次数 exetcpuDataset.createOrReplaceTempView("gbupload_etctu") val gbupload_etctu: Dataset[GBUPLoad] = spark.sql( """ |select vehicleid as vehicleid, |(case | when 0 <= hour(transtime) and hour(transtime)<=3 then 'A' | when 3 < hour(transtime) and hour(transtime)<=6 then 'B' | when 6 < hour(transtime) and hour(transtime)<=9 then 'C' | when 9 < hour(transtime) and hour(transtime)<=12 then 'D' | when 12 < hour(transtime) and hour(transtime)<=15 then 'E' | when 15 < hour(transtime) and hour(transtime)<=18 then 'F' | when 18 < hour(transtime) and hour(transtime)<=21 then 'G' | when 21 < hour(transtime) and hour(transtime)<24 then 'H' | else 'I' end) as transtime, |day |from |gbupload_etctu |""".stripMargin).as[GBUPLoad] gbupload_etctu.persist(StorageLevel.MEMORY_AND_DISK) gbupload_etctu.checkpoint() gbupload_etctu.unpersist() spark.sqlContext.dropTempTable("gbupload_etctu") //todo：根据车牌号进行分组聚合，统计最近一次出行时间、出行总时长、最长出行时间、出行总里程、出行花费总金额、负重出行总里程、出行次数 //合并etc出口和现金出口的数据 val trcDataFrame = trcDataSet_p .groupBy($"day", $"vehicleid") .agg(immutable.Map("exTime" -> "max", "travel_time" -> "sum", "travel_time" -> "max", "feemileage" -> "sum", "fee" -> "sum", "weight_mileage" -> "sum", "vehicleid" -> "count")) .withColumnRenamed("max(exTime)","max_exTime") .withColumnRenamed("max(travel_time)","max_travel_time") .withColumnRenamed("sum(travel_time)","sum_travel_time") .withColumnRenamed("sum(feemileage)","sum_feemileage") .withColumnRenamed("sum(fee)","sum_fee") .withColumnRenamed("sum(weight_mileage)","sum_weight_mileage") .withColumnRenamed("count(vehicleid)","exTimes_count") //todo： 取车辆轴数和收费类型 trcDataSet_p.createOrReplaceTempView("trc_table") val vehicleTypeAxlecount: DataFrame = spark.sql( """ |select |a.day as day, |a.vehicleid as vehicleid, |a.axlecount as axlecount, |a.vehicletype as vehicletype |from |(select day , |vehicleid , | axlecount , | vehicletype , | row_number() over(partition by day , vehicleid order by exTime desc) as rr |from |trc_table) a where a.rr=1 |""".stripMargin) // val trc_C_V_DataFrame: DataFrame = trcDataFrame.join(vehicleTypeAxlecount.select($"axlecount",$"vehicletype",$"day",$"vehicleid"), (trcDataFrame.col("vehicleid") === vehicleTypeAxlecount.col("vehicleid")).and(trcDataFrame.col("day") === vehicleTypeAxlecount.col("day")), "left") // 合并出口统计数据与用户轴数 val trc_C_V_DataFrame: DataFrame = trcDataFrame.join(vehicleTypeAxlecount, Seq("day","vehicleid"), "left") //todo: 抽取出口数据集中的省份编码 val trcProvinceDataSet: Dataset[Row] = trcDataSet_p.select($"day", $"vehicleid", $"travel_provinces").groupBy($"day", $"vehicleid", $"travel_provinces").count().withColumnRenamed("count","travel_provinces_count") trcProvinceDataSet.createOrReplaceTempView("trcProvince_table") val trcProvinceCountDataSet = spark.sql( """ |select day , vehicleid as vehicleid ,concat_ws("#",collect_list(concat_ws("%",travel_provinces,travel_provinces_count) )) as travel_provinces_count |from trcProvince_table |group by day , vehicleid |""".stripMargin) val trcTransProvinceCountDataFrame = trc_C_V_DataFrame.join(trcProvinceCountDataSet, Seq("day","vehicleid"), "outer") //统计每个牌照出行时间段的次数 val transtimeDataFrame:DataFrame = gbupload_etctu.groupBy($"day",$"vehicleid",$"transtime") .agg(count("transtime").as("transtime_preferences_count")) // 汇总车辆各个时间段出行次数，变成了map结构 transtimeDataFrame.createOrReplaceTempView("transtimeData") val transTimeCountMapDataFrame: DataFrame = spark.sql( """ |select day , |vehicleid , |concat_ws("#",collect_list(concat_ws("%",transtime,transtime_preferences_count) )) as transtime_count |from transtimeData |group by day , vehicleid |""".stripMargin) var trcResultDataFrame = trcTransProvinceCountDataFrame.join(transTimeCountMapDataFrame, Seq("day","vehicleid"), "outer") spark.sqlContext.dropTempTable("trcDataSet") spark.sqlContext.dropTempTable("trcProvince_table") spark.sqlContext.dropTempTable("transtimeData") spark.sqlContext.dropTempTable("gbupload_etctu") spark.sqlContext.dropTempTable("trc_table") log.info(trcResultDataFrame.printSchema()) if (result_partitions> 0){ trcResultDataFrame= trcResultDataFrame.repartition(result_partitions) } trcResultDataFrame.write.csv(out_path) spark.stop() } def formatSparkSession(): SparkSession ={ // todo: 实例化SparkSession val spark: SparkSession = SparkSession .builder() // .appName("DataExtract") // .master("local[*]") .getOrCreate() spark } def exists(path:String):Boolean={ Files.exists(Paths.get(path)) } }