1, 设备安装app统计分析

代码有删减,但整体保持原结构

  1. package com.wsy.learn
  3. import org.apache.hadoop.conf.Configuration
  4. import org.apache.hadoop.fs.{FileSystem, Path}
  5. import org.apache.spark.sql.SparkSession
  6. import org.apache.spark.sql.functions._
  7. import org.apache.spark.internal.Logging
  8. import org.apache.spark.storage.StorageLevel
  10. object InstallAppCount extends Logging {
  12.     def main(args: Array[String]): Unit = {
  13.         if (args.length > 0) {
  14.             logInfo(s"------ args=${args(0)}----------")
  15.         } else {
  16.             logError(s"------ 请传递一个逗号分割的查询开始日期,结束日志,结果保存路径的参数,例如 20200615,20200616,/your/path ----------")
  17.             System.exit(1)
  18.         }
  20.         val startDate = args(0).split(",")(0).trim
  21.         val endDate = args(0).split(",")(1).trim
  22.         val output = args(0).split(",")(2).trim
  23.         logInfo(s" -------  startDate=${startDate}, endDate=${endDate}, output=${output}  ----------")
  26.         val spark = SparkSession.builder()
  27.                 .master("yarn")
  28.                 .appName("install_app_count")
  29.                 .config("spark.sql.strict.mode", false)
  30.                 .config("spark.default.parallelism", 200)
  31.                 .config("spark.sql.shuffle.partitions", 200)
  32.                 .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  33.                 .enableHiveSupport()
  34.                 .getOrCreate()
  37.         //spark.sqlContext.setConf("hive.exec.dynamic.partition", "true")
  38.         //spark.sqlContext.setConf("hive.exec.dynamic.partition.mode", "nonstrict")
  40.         //此路径可以已存在
  41.         spark.sparkContext.setCheckpointDir("/home/wsy/dir")
  44.         import spark.implicits._
  45.         //spark.sql("set spark.sql.strict.mode = false ")
  47.         spark.sql("use your_database")
  48.         //table 是分区表,select必须在where中指定分区
  49.         val df1 = spark.sql(s"select module, x from your_table where (modulename = 'a' or modulename = 'b' )  and day >= '${startDate}' and day < '${endDate}' ")
  53.         val df2 = df1.select(
  54.             $"module",
  55.             get_json_object($"x", "$.a").as("a"),
  56.             get_json_object($"x", "$.b").as("b"),
  57.             when($"module" === "a", get_json_object($"x", "$.a"))
  58.                     .otherwise(get_json_object($"x", "$.b"))
  59.                     .as("device")
  60.         )
  61.         val df3 = df2.filter($"x" === "3" || $"x" === "4").filter($"device".isNotNull).repartition(200)
  64.         val df4 = df3.withColumn("deviceid",
  65.             when($"x" === "3", get_json_object($"device", "$.a"))
  66.                     .otherwise(get_json_object($"device", "$.b"))
  67.         )
  68.         val df5 = df4.select($"module", $"a", $"b", $"c", $"deviceid")
  69.         //优化复用,原始数据一个月3.5T, df5大概220GB,对3.5Tshuffle,250core大概1小时,500core,大概30分钟
  70.       //  persist或者checkpoint,把原来3小时40分钟缩短到40分钟,原程序对df5有5次复用,这里代码有删减
  71.       // 缓存以后,原来重复5遍的计算,只需要一次,是一个大大的优化
  72.       df5.persist(StorageLevel.MEMORY_AND_DISK_SER)
  73.         //val df5=df51.checkpoint()
  74.         df5.count()
  78.         val df_app = df5.filter($"source" === "ios" || $"source" === "android")
  79.                 .select($"source", $"deviceid")
  80.                 .distinct()
  81.                 .repartition(100)
  82.         df_app.cache()
  83.         val result_app = df_app.count()
  85.         val df_accountsystem = df5.filter($"tixiid" === "1")
  86.                 .select($"tixiid", $"deviceid", $"source")
  87.                 .distinct()
  88.                 .repartition(100)
  89.         df_accountsystem.cache()
  90.         val result_accountsystem = df_accountsystem.count()
  92.         val result_x: Seq[String] = Seq(s"安装APP的数量, ${result_app}", s"安装账号体系数量, ${result_accountsystem}")
  94.               val df_x = df5.filter($"a" === "ios" || $"a" === "android")
  95.                 .select($"a", $"deviceid")
  96.                 .distinct()
  97.                 .repartition(100)
  98.         df_x.cache()
  99.         val result_11: Long = df_x.count()
  101.         //统计结果
  103.         val df_accountsystem_1 = df_accountsystem.filter($"a" =!= "ios" && $"a" =!= "android")
  104.                 .select($"x", $"deviceid")
  106.         val result_12: Long = df_x.join(df_app, "deviceid").count()
  107.         val result_13: Long = df_x.join(df_accountsystem_1, "deviceid").count()
  109.         val result_1: Seq[String] = Seq(s"安装x APP的数量, ${result_11}", s"同时安装x及y APP的数量, ${result_12}", s"同时安装x 及账号体系其他APP的数量, ${result_13}")
  115.         //如果已存在,删除,防止目录已存在异常
  116.         val conf = new Configuration()
  117.         val fs = FileSystem.get(conf)
  118.         val output_path = new Path(output)
  119.         if (fs.exists(output_path)) {
  120.             fs.delete(output_path, true)
  121.         }
  124.         val result: Seq[String] = result_x ++ result_1
  125.         logInfo(s"------result=${result}---------------")
  126.         spark.sparkContext.parallelize(result, 1).saveAsTextFile(output)
  128.         spark.stop()
  131.     }
  133. }