Merge branch 'master' of gitlab.mobvista.com:dataplatfrom/mobvista-dmp

azkaban/ali/reyun/reyun_label_baijiu.job

+type=command
+command=sh -x reyun_label_baijiu.sh
\ No newline at end of file

azkaban/ali/reyun/reyun_label_baijiu.sh

+#!/bin/sh
+
+# # # # # # # # # # # # # # # # # # # # # #
+# @author : jiangfan
+# @date : 2021-07-01 12:06:00
+# # # # # # # # # # # # # # # # # # # # # #
+#!/usr/bin/env bash
+
+source ../../dmp_env.sh
+
+dt_today=$(date -d "$ScheduleTime 1 days ago" +"%Y%m%d")
+dt_slash_today=$(date -d "$ScheduleTime 1 days ago" +"%Y/%m/%d")
+
+
+OUTPUT_PATH="${REYUN_LABEL_BAIJIU_DAILY_PATH}/${dt_slash_today}"
+
+hadoop fs -rm -r "${OUTPUT_PATH}"
+
+spark-submit --class mobvista.dmp.datasource.reyun.ReyunLabelBaijiu \
+ --conf spark.yarn.executor.memoryOverhead=2048  \
+ --conf spark.network.timeout=720s \
+ --conf spark.driver.maxResultSize=4g \
+ --conf spark.default.parallelism=2000 \
+ --conf spark.sql.shuffle.partitions=2000 \
+ --conf spark.sql.broadcastTimeout=1200 \
+ --conf spark.sql.autoBroadcastJoinThreshold=31457280 \
+ --files ${HIVE_SITE_PATH} \
+ --jars ${JARS} \
+ --master yarn --deploy-mode cluster  --executor-memory 8g --driver-memory 4g  --executor-cores 4  --num-executors 170 \
+ ../../${JAR}  \
+ -output ${OUTPUT_PATH} -coalesce 680  -dt_today ${dt_today}
+
+if [[ $? -ne 0 ]];then
+  exit 255
+fi
+

azkaban/dmp_env.sh

@@ -358,6 +358,7 @@ REYUN_RAW_DATA="s3://mob-emr-test/reyun/pkginfo"
 
 # reyun business tmp data
 REYUN_LABEL_TEST_DAILY_PATH="s3://mob-emr-test/dataplatform/DataWareHouse/data/dwh/tmp/reyun_label_test_daily"
+REYUN_LABEL_BAIJIU_DAILY_PATH="s3://mob-emr-test/dataplatform/DataWareHouse/data/dwh/tmp/reyun_label_baijiu_daily"
 
 # alipay_activation business tmp data
 ALIPAY_ACTIVATION_DAILY_PATH="s3://mob-emr-test/dataplatform/DataWareHouse/data/dwh/etl_alipay_activation_daily"

src/main/scala/mobvista/dmp/datasource/reyun/ReyunLabelBaijiu.scala

+package mobvista.dmp.datasource.reyun
+
+import mobvista.dmp.common.CommonSparkJob
+import mobvista.dmp.util.DateUtil
+import org.apache.commons.cli.Options
+import org.apache.hadoop.fs.{FileSystem, Path}
+import org.apache.spark.sql.SparkSession
+
+import java.net.URI
+
+/**
+ * @author jiangfan
+ * @date 2021/7/23 18:00
+ */
+
+class ReyunLabelBaijiu  extends CommonSparkJob with Serializable{
+  override protected def buildOptions(): Options = {
+    val options = new Options
+    options.addOption("coalesce", true, "[must] coalesce")
+    options.addOption("output", true, "[must] output")
+    options.addOption("dt_today", true, "[must] dt_today")
+    options
+  }
+
+  override protected def run(args: Array[String]): Int = {
+    val commandLine = commParser.parse(options, args)
+    if (!checkMustOption(commandLine)) {
+      printUsage(options)
+      return -1
+    } else printOptions(commandLine)
+
+    val coalesce = commandLine.getOptionValue("coalesce")
+    val output = commandLine.getOptionValue("output")
+    val dt_today = commandLine.getOptionValue("dt_today")
+
+    val spark = SparkSession.builder()
+      .appName("ReyunLabelBaijiu")
+      .config("spark.rdd.compress", "true")
+      .config("spark.io.compression.codec", "snappy")
+      .config("spark.sql.orc.filterPushdown", "true")
+      .config("spark.sql.warehouse.dir", "s3://mob-emr-test/spark-warehouse")
+      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
+      .enableHiveSupport()
+      .getOrCreate()
+
+    val sc = spark.sparkContext
+    import spark.implicits._
+
+    FileSystem.get(new URI(s"s3://mob-emr-test"), spark.sparkContext.hadoopConfiguration).delete(new Path(output), true)
+
+    val one_day_ago = DateUtil.getDayByString(dt_today, "yyyyMMdd", -1)
+    val two_days_ago = DateUtil.getDayByString(dt_today, "yyyyMMdd", -2)
+
+    try {
+      val sql1=
+        s"""
+           |select distinct device_id
+           |from
+           |(select md5(imei) as device_id
+           |from
+           |(select app_id,
+           |(case when imei not in ('00000000-0000-0000-0000-000000000000','0000000000000000','0','',' ')  then imei else null end) as imei,
+           |(case when ext_oaid not in ('00000000-0000-0000-0000-000000000000','0000000000000000','0','',' ')  then ext_oaid else null end ) as oaid
+           |from
+           |dwh.ods_adn_trackingnew_request
+           |where concat(yyyy,mm,dd) in ('${dt_today}')
+           |and country_code = 'CN'
+           |and platform = 'android'
+           |union all
+           |select app_id,(case when imei not in ('00000000-0000-0000-0000-000000000000','0000000000000000','0','',' ')  then imei else null end) as imei,
+           |(case when ext_oaid not in ('00000000-0000-0000-0000-000000000000','0000000000000000','0','',' ')  then ext_oaid else null end ) as oaid
+           |from dwh.ods_adn_trackingnew_hb_request
+           |where concat(yyyy,mm,dd)  in ('${dt_today}')
+           |and country_code = 'CN'
+           |and platform = 'android'
+           |) as tmp
+           |where  tmp.imei is not null  and tmp.app_id in (select appid from dwh.reyun_label_baijiu_filter_appid)
+           |union all
+           |select md5(oaid) as device_id
+           |from
+           |(select app_id,
+           |(case when imei not in ('00000000-0000-0000-0000-000000000000','0000000000000000','0','',' ')  then imei else null end) as imei,
+           |(case when ext_oaid not in ('00000000-0000-0000-0000-000000000000','0000000000000000','0','',' ')  then ext_oaid else null end ) as oaid
+           |from
+           |dwh.ods_adn_trackingnew_request
+           |where concat(yyyy,mm,dd)  in ('${dt_today}')
+           |and country_code = 'CN'
+           |and platform = 'android'
+           |union all
+           |select app_id,(case when imei not in ('00000000-0000-0000-0000-000000000000','0000000000000000','0','',' ')  then imei else null end) as imei,
+           |(case when ext_oaid not in ('00000000-0000-0000-0000-000000000000','0000000000000000','0','',' ')  then ext_oaid else null end ) as oaid
+           |from dwh.ods_adn_trackingnew_hb_request
+           |where concat(yyyy,mm,dd) in ('${dt_today}')
+           |and country_code = 'CN'
+           |and platform = 'android'
+           |) as tmp
+           |where  tmp.oaid is not null and tmp.app_id in (select appid from dwh.reyun_label_baijiu_filter_appid)
+           |union all
+           |select if(imeimd5 is null ,md5(imei),imeimd5)  as device_id
+           |from
+           |(select  split(ext5,',')[4] as imei,split(ext5,',')[5] as imeimd5
+           |from adn_dsp.log_adn_dsp_request_orc_hour
+           |where concat(yr,mt,dt)  in ('${dt_today}')  and hh >='00'  and hh<'06'
+           |and countrycode = 'CN'
+           |and os = 'android'
+           |and exchanges='xunfei'
+           |) adx
+           |union all
+           |select if(oaidmd5 is null ,md5(oaid),oaidmd5)  as device_id
+           |from
+           |(select  split(ext5,',')[12] as oaid ,split(ext5,',')[13] as oaidmd5
+           |from adn_dsp.log_adn_dsp_request_orc_hour
+           |where concat(yr,mt,dt)  in ('${dt_today}')  and hh >='00'  and hh<'06'
+           |and countrycode = 'CN'
+           |and os = 'android'
+           |and exchanges='xunfei'
+           |) adx
+           |) tmp
+           |""".stripMargin
+
+      println("sql=============="+sql1)
+
+      spark.sql(sql1).rdd.map(_.mkString).coalesce(coalesce.toInt).saveAsTextFile(output)
+
+    } finally {
+      spark.stop()
+    }
+    0
+  }
+}
+
+object ReyunLabelBaijiu {
+  def main(args: Array[String]): Unit = {
+     new ReyunLabelBaijiu().run(args)
+  }
+}