数据过滤

电玩女神 2022-03-31 04:48 237阅读 0赞

package emg.branchs
    
    import java.sql.{Connection, DriverManager}
    
    import emg.util._
    import org.apache.spark.SparkConf
    import org.apache.spark.sql.SparkSession
    
    import scala.collection.mutable.ArrayBuffer
    
    /**
      * @Auther: sss
      * @Date: 2018/12/26 18:42
      * @Description:
      */
    object EmgFilters {
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName(this.getClass.getSimpleName)
          //.setMaster("local[*]")
         .setMaster("spark://192.168.00.00:7077")
        val session = SparkSession.builder()
          .config(conf)
          .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //使用kryo序列化的类库
          /*     .config("spark.shuffle.manager", "hash")//设置为hashShuffleManager
               .config("spark.shuffle.consolidateFiles", "true")//开启consolidate机制*/
          .config("spark.default.parallelism", "50") //设置并行度
          .config("spark.shuffle.file.buffer", "64k") //设置map端内存缓冲区大小
          .config("spark.reducer.maxSizeInFlight", "96M") //设置reduce端拉取数据的大小
          .config("spark.shuffle.memoryFraction", "0.3") //reduce端聚合内存占比
          .config("spark.shuffle.io.maxRetries", "6") //最多拉取次数
          .getOrCreate()
    
    
        val url = "jdbc:hive2://192.168.40.51:10000/emg"
        val username = "root"
        val password = "123456"
    
        val driverName = "org.apache.hive.jdbc.HiveDriver"
        try {
          Class.forName(driverName)
        } catch {
          case e: ClassNotFoundException =>
            println("Missing Class", e)
        }
        val con: Connection = DriverManager.getConnection(url, username, password)
        val state = con.createStatement()
    
        import session.implicits._
        // logger.info("")
    
        // hdfs://192.168.40.51:9000/user/hive/warehouse/emg.db/d181119/*    郑州一天的数据
    
        var inputFileName = ""
        var ouputFileName = ""
        val time = Time.getFrontHour().split("-")
        //读取上一个小时的数据 Time.getFrontHour()
        inputFileName = "hdfs://192.168.40.51:9000/user/emg/ecar/" + time(0) + time(1) + "/" + time(2) + "/" + time(3) + "/"
        //按小时输出目录
        ouputFileName = "hdfs://192.168.40.51:9000/user/emg/cxb_out/out/" + time(0) + time(1) + "/" + time(2) + "/" + time(3) + "/"
    
        println("读取路径为：" + inputFileName)
        println("输出路径为：" + ouputFileName)
    
    
        val source = session.read.textFile(inputFileName)
    
        //给源数据赋上schema
        val value = source.map(f => {
          val uuid = f.split(",", f.length)(0)
          val travel_type = f.split(",", f.length)(1).toInt
          val location_type = f.split(",", f.length)(2).toInt
          val lat = f.split(",", f.length)(3).toDouble
          val lon = f.split(",", f.length)(4).toDouble
          val altitude = f.split(",", f.length)(5).toDouble
          val precision = f.split(",", f.length)(6).toDouble
          val dir = f.split(",", f.length)(7).toDouble
          val speed = f.split(",", f.length)(8).toDouble
          val timestamp = f.split(",", f.length)(9).toLong
          val load_status = f.split(",", f.length)(10).toInt
          val valid = f.split(",", f.length)(11).toInt
          Emgfilter(uuid, travel_type, location_type, lat, lon, altitude, precision, dir, speed, timestamp, load_status, valid)
        })
        value.createOrReplaceTempView("tmp")
        //创建表过滤出所需字段
        val result01 = session.sql("select uuid,lat,lon,timestamp  from tmp")
    
        //3410111
        val result02 = result01.toDF().rdd.map(row => {
          val uuid = row.get(0).toString
          val lat = row.get(1).toString.toDouble
          val lon = row.get(2).toString.toDouble
          val utc = row.get(3).toString.toLong
          MyData1(uuid, lat, lon, utc)
        })
    
        val res3 = result02.map(d => {
          (d.uuid, d)
        }).groupByKey().mapValues(datas => {
          datas.toList.sortBy(d => d.utc)(Ordering[Long])
        })
          .mapValues(datas => {
            var tmp1 = MyData1("", 0, 0, 0)
            var result = new ArrayBuffer[ArrayBuffer[MyData2]]
            var flag = true
            //判断是否是第一次循环
            var i = 0 // 标识轨迹段的起始
            var count = 0
            //循环计数 测试用，可删除
            var result_crr = new ArrayBuffer[MyData2]
            //当前轨迹段
            var count2 = 0.0 //用于累加计算当前轨迹的长度 ，用于判断轨迹长度是否大于300
            /**
              * 循环当前用户的所有数据
              */
            for (data <- datas) {
              count += 1
              //循环计数+1
              var now = MyData2("", 0, 0, 0, -1) //用于存放当前数据点
              if (flag) {
                tmp1 = data //存放当前点，用于与后面的点比较
                now = MyData2(data.uuid, data.lat, data.lon, data.utc, 0) //存放第一个点做为当前轨迹段的起始点，用0标记
                result_crr += now
                flag = false //将flag设置为false，循环不在进入此代码块
              } else {
                val distance = Test.getDistance(tmp1.lon, tmp1.lat, data.lon, data.lat)
                //获取当前点与上一个点的时间差取绝对值
                val time = Math.abs(data.utc - tmp1.utc)
                //调试查看
                /* if (count < 20) {
                   println(data + "  " + tmp1 + "  " + distance + "  " + time)
                 }*/
                //如果符合普通点
                if (!(distance < 50 || time < 10)) {
                  //再判断是否是断点大于200且大于30秒，则打断，当前轨迹结束
                  if (distance > 200 || time > 30) {
                    now = MyData2(data.uuid, data.lat, data.lon, data.utc, 0)
                    if (result_crr.size >= 2 && count2 >= 300) {
                      result += result_crr
                    } else {
                      //不合格则丢弃，此处else块用于调试，可删除
                      /*if (count < 50) {
                        println("被抛弃的轨迹段：")
                        println(result_crr) //打印被抛弃的轨迹
                        println("----------轨迹段结束----------")
                      }*/
                    }
                    result_crr = new ArrayBuffer[MyData2] //给result_crr重新赋值，相当于清空，创建一个新的轨迹段
                    result_crr += now
                    tmp1 = data
                  } else {
                    //小于200直接添加，-1表示普通点
                    now = MyData2(data.uuid, data.lat, data.lon, data.utc, -1) //存放当前点判断之后的结果，非断点必然是普通点，用-1标识
                    result_crr += now
                    count2 += distance //将当前点与上一个点的距离累加进当前轨迹段的距离总和
                    tmp1 = data //存放当前点，用于与后面的点比较
                  }
                }
              }
            }
            result.flatten //返回结果 并压平
          })
    
        val res4 = res3.map(_._2).filter(_.size > 0)
          .map(t => t.map(t => t.uuid + "\t" + t.lat + "\t" + t.lon + "\t" + t.utc + "\t" + t.log))
          .map(t => t.mkString("\n"))
          // .coalesce(1)
          .saveAsTextFile(ouputFileName)
    
        println("===============================保存成功==========================")
    
    
        var partitions = time(0) + time(1) + time(2) + time(3)
    
        var paths = "/user/emg/cxb_out/out/" + time(0) + time(1) + "/" + time(2) + "/" + time(3) + "/"
    
        //由于hive的元数据库启动不了 连接不上  只能用jdbc的方式将结果load进hive表中
        var sql2 = "load data  inpath '" + paths + "' into table finallyResult partition (ds ='" + partitions + "')"
    
        try {
          val assertion = state.execute(sql2)
          state.execute(sql2)
          println("===============================存入hvie成功==========================")
        } catch {
          case e: Exception => e.printStackTrace()
        } finally {
          if (null != con) {
            con.close()
          }
        }
        session.close()
      }
    }