1、产品概述

大数据设计的DMP标签体系致力于对创蓝所用于的号码用户群，进行标签体系构建，包括触达标签，偏好标签，私域流量标签，基础信息标签等，通过不同来源数据整合，清洗，挖掘构建一个公司层面的用户标签体系。当前整个DMP标签体系正在设计开发中。
origin_img_v2_6a749ba4-ab2f-4838-b0d1-8285de99710g.png

2、触达标签的大表格的设计

mobile(号码) telcom （运营商） base_province （号码省份） base_city （号码城市） appplatform（平台1:IOS 2:安卓） devicename （设备） imei（设备信息） oaid（设备信息） aim(aim方式) aim_device（aim设备） rcs_video(视频短信) rcs_5g(5g触达) rcs_tailsms(贴尾短信) rcs_ip(ip短信) rcs_number_porting(携号转网) rcs_status（号码状态） …

3、表格构建代码

  1. CREATE TABLE `cl_nlp.nlp_contact_label_system`(
  2.   `mobile` string COMMENT '手机号',
  3.   `telcom` string COMMENT '运营商',
  4.   `base_province` string COMMENT '手机号归属省份',
  5.   `base_city` string COMMENT '手机号归属市',
  6.   `applatform` string COMMENT '平台1:IOS 2:安卓',
  7.   `devicename` string COMMENT '设备名称',
  8.   `imei` string COMMENT '设别信息imei',
  9.   `oaid` string COMMENT '设备信息oaid',
  10.   `aim` string COMMENT 'aim方式1:aim触达，0：未知',
  11.   `aim_device` string COMMENT 'aim设备',
  12.   `rcs_video` string COMMENT '1:视频短信，0：未知',
  13.   `rcs_5g` string COMMENT '1:5g触达，0：未知',
  14.   `rcs_tailsms` string COMMENT '1:贴尾短信，0：未知',
  15.   `rcs_ip` string COMMENT '1：ip短信，0：未知',
  16.   `rcs_number_porting` string COMMENT '1：携号转网，0：未知',
  17.   `rcs_status` string COMMENT '号码状态'
  18. )
  19. PARTITIONED BY (
  20.   `ptt_day` string)
  21. ROW FORMAT SERDE
  22.   'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
  23. STORED AS INPUTFORMAT
  24.   'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
  25. OUTPUTFORMAT
  26.   'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
  27. TBLPROPERTIES("parquet.compression"="SNAPPY");

注：生产环境中，这张表还没有建立，因为后续可能融入偏好和私域的标签，表格可能需要扩列。

4、触达标签数据流

origin_img_v2_569af1c5-14f9-4432-ac3d-291465d9278g.png

4.1 创蓝号码池

  • cl_nlp.nlp_mobile_di

由于数据量较大，需要规范一定的范围，缩小号码量，作为dmp标签的号码池，该号码池可以每月（或其他）更新一次。

  1. spark.sql(
  2.       """
  3.         |select
  4.         |    mobile
  5.         |from
  6.         |    cl_nlp.nlp_mobile_di
  7.         |where
  8.         |     max_date>'$time'
  9.         |group by
  10.         |     mobile
  11.         |""".stripMargin).cache().createOrReplaceTempView("tmp1_mobile")

4.2 aim方式

当前可实现aim的有三种类型机型的表格：

  • cl_nlp.nlp_oppo_aim_df:7690
  • cl_nlp.nlp_device_xiaomi_aim_df:40122683
  • cl_nlp.nlp_aliyun_aim_df

这三张表中分别包含mobile（号码）和device（设备）字段
这三张表的数据量较少，先将三张表union all

  1. spark.sql(
  2.       """
  3.         |select
  4.         |    substr(mobile,4) mobile,
  5.         |    device
  6.         |from
  7.         |    cl_nlp.nlp_oppo_aim_df
  8.         |union all
  9.         |select
  10.         |    mobile,
  11.         |    device
  12.         |from
  13.         |    cl_nlp.nlp_device_xiaomi_aim_df
  14.         |group by
  15.         |     mobile,
  16.         |     device
  17.         |union all
  18.         |select
  19.         |    mobile,
  20.         |    device
  21.         |from
  22.         |    cl_nlp.nlp_aliyun_aim_df
  23.         |group by
  24.         |    mobile
  25.         |    device
  26.         |""".stripMargin).cache().createOrReplaceTempView("tmp2_aim")

4.3 ip短信

  • cl_nlp.dwd_vivo_device_result_agg

该表格是以日全量量进行更新，所以获取最新日期的数据作为有效的号码

  1.  spark.sql(
  2.       s"""
  3.         |select
  4.         |    mobile
  5.         |from
  6.         |    cl_nlp.dwd_vivo_device_result_ag
  7.         |where
  8.         |    ptt_day='$ptt_day'
  9.         |group by
  10.         |    mobile
  11.         |""".stripMargin).cache().createOrReplaceTempView("tmp3_ip")

4.4 视频短信

  • cl_ods.ods_rcs_m_batch_detail_di

该表格中号码进行加密，获取有效号码时需要先解密

  1. spark.sql(
  2.   s"""
  3.     |select
  4.     |    concat(cl_ods.decrypt(substr(phone,1,30)),substr(phone,31,34)) mobile,
  5.     |from
  6.     |    cl_ods.ods_rcs_m_batch_detail_di
  7.     |group by
  8.     |    mobile
  9.     |""".stripMargin).cache().createOrReplaceTempView("tmp2_video")

4.5 SDK设备信息

  • cl_cdm.dwd_fls_sdk_mobile_detail_org_di

从SDK表格数据中可以获取运营商，手机归属省份相关信息，平台，设备名，imei设备信息，oaid设备信息等信息，由于该表是依据用户点击时间记录，所以使用用户最新一条数据代表用户信息。

  1. spark.sql(
  2.       s"""
  3.         |select
  4.         |    *
  5.         |from
  6.         |    (select
  7.         |        mobile,
  8.         |        telcom,
  9.         |        base_province,
  10.         |        base_city,
  11.         |        appplatform,
  12.         |        devicename,
  13.         |        imei,
  14.         |        oaid,
  15.         |        row_number()over(partition BY mobile ORDER BY m_time DESC) rn
  16.         |    from
  17.         |        cl_cdm.dwd_fls_sdk_mobile_detail_org_di
  18.         |    where
  19.                  ptt_mon<="$ptt_mon"
  20.         |     ) t
  21.         |where
  22.         |   t.rn = 1
  23.         |""".stripMargin).cache().createOrReplaceTempView("tmp3_sdk")

4.6 贴尾触达

注：好像是依据国内短信的批次号来识别（目前还不明确）

4.7 5G触达

当前逻辑关系还不明确

4.8 普通短信

  • 运营商
  • 携号转网
  • 号码状态

这部分的表格还在处理中

5、整体撞库拼接代码

  1. package com.bigdata.Temporary.tmp
  3. import org.apache.spark.sql.SparkSession
  5. import java.text.SimpleDateFormat
  6. import java.util.Date
  8. object DMP_CD {
  10.   def main(args: Array[String]): Unit = {
  11.     val spark = SparkSession.builder()
  12.       .appName("DMP_CD_Day")
  13.       .config("hive.exec.dynamic.partition", "true")
  14.       .config("hive.exec.dynamic.partition.mode", "nonstrict")
  15.       .config("hive.warehouse.subdir.inherit.perms", "false")
  16.       .enableHiveSupport()
  17.       .getOrCreate()
  19.     // 定义变量，日更的时间，月更的时间
  20.     val sf1 = new SimpleDateFormat("yyyyMMdd")
  21.     //当前的时间多一天的时间
  22.     val date1 = new Date(System.currentTimeMillis - 1000 * 60 * 60 * 24)
  23.     val ptt_day = sf1.format(date1)
  26.     // 构建有效号码池
  27.     spark.sql(
  28.       """
  29.         |select
  30.         |    mobile
  31.         |from
  32.         |    cl_nlp.nlp_mobile_di
  33.         |where
  34.         |     max_date>'20220101'
  35.         |group by
  36.         |     mobile
  37.         |""".stripMargin).cache().createOrReplaceTempView("tmp1_mobile")
  39.     // 将aim方式三张数据表拼接到一起
  40.     // 按照日更新的数据为：cl_nlp.nlp_oppo_aim_df,cl_nlp.nlp_xiaomi_aim_df,cl_nlp.nlp_aliyun_aim_df
  41.     spark.sql(
  42.       """
  43.         |select
  44.         |    substr(mobile,4) mobile,
  45.         |    device
  46.         |from
  47.         |    cl_nlp.nlp_oppo_aim_df
  48.         |union all
  49.         |select
  50.         |    mobile,
  51.         |    device
  52.         |from
  53.         |    cl_nlp.nlp_device_xiaomi_aim_df
  54.         |group by
  55.         |     mobile,
  56.         |     device
  57.         |union all
  58.         |select
  59.         |    mobile,
  60.         |    device
  61.         |from
  62.         |    cl_nlp.nlp_aliyun_aim_df
  63.         |group by
  64.         |    mobile
  65.         |    device
  66.         |""".stripMargin).cache().createOrReplaceTempView("tmp2_aim")
  70.     //ip短信：cl_nlp.dwd_vivo_device_result_agg,天全量表
  71.     spark.sql(
  72.       s"""
  73.         |select
  74.         |    mobile
  75.         |from
  76.         |    cl_nlp.dwd_vivo_device_result_agg
  77.         |where
  78.         |    ptt_day='$ptt_day'
  79.         |group by
  80.         |    mobile
  81.         |""".stripMargin).cache().createOrReplaceTempView("tmp3_ip")
  84.     // 数据量较大:去重后亿级别
  85.     // 视频短信：cl_ods.ods_rcs_m_batch_detail_di
  86.     spark.sql(
  87.       s"""
  88.          |select
  89.          |    concat(cl_ods.decrypt(substr(phone,1,30)),substr(phone,31,34)) mobile,
  90.          |from
  91.          |    cl_ods.ods_rcs_m_batch_detail_di
  92.          |group by
  93.          |    mobile
  94.          |""".stripMargin).cache().createOrReplaceTempView("tmp4_video")
  96.     // 数据量较大，去重后亿级别
  97.     //SDK数据：cl_cdm.dwd_fls_sdk_mobile_detail_org_di
  98.     spark.sql(
  99.       s"""
  100.          |select
  101.          |    *
  102.          |from
  103.          |    (select
  104.          |        mobile,
  105.          |        telcom,
  106.          |        base_province,
  107.          |        base_city,
  108.          |        appplatform,
  109.          |        devicename,
  110.          |        imei,
  111.          |        oaid,
  112.          |        row_number()over(partition BY mobile ORDER BY m_time DESC) rn
  113.          |    from
  114.          |        cl_cdm.dwd_fls_sdk_mobile_detail_org_di
  115.          |     ) t
  116.          |where
  117.          |   t.rn = 1
  118.          |""".stripMargin).cache().createOrReplaceTempView("tmp5_sdk")
  120.     // 表格与号码池之间拼接
  121.     //aim数据与号码库拼接
  122.     spark.sql(
  123.       """
  124.         |select
  125.         |    tmp1_mobile.mobile
  126.         |    tmp2_aim.device
  127.         |    if(tmp2.device is NULL,0,1) as aim
  128.         |from
  129.         |    tmp1_mobile
  130.         |left join
  131.         |    tmp2_aim
  132.         |on
  133.         |    tmp1_mobile.mobile=tmp2_aim.mobile
  134.         |""".stripMargin).cache().createOrReplaceTempView("tmp6_aim_mobile")
  136.     // ip号码与号码库拼接
  137.     spark.sql(
  138.       """
  139.         |select
  140.         |    tmp6_aim_mobil.mobile,
  141.         |    tmp6_aim_mobi.device
  142.         |    tmp6_aim_mobi.aim
  143.         |    if(tmp3_ip is null,0,1) as rcs_ip
  144.         |from
  145.         |    tmp6_aim_mobile
  146.         |left join
  147.         |    tmp3_ip
  148.         |on
  149.         |    tmp6_aim_mobile.mobile = tmp3_ip.mobile
  150.         |""".stripMargin).cache().createOrReplaceTempView("tmp7_aim_ip_mobile")
  152.     //视频短信号码与号码库拼接
  153.     spark.sql(
  154.       """
  155.         |select
  156.         |    tmp7_aim_ip_mobile.mobile,
  157.         |    tmp7_aim_ip_mobile.device,
  158.         |    tmp7_aim_ip_mobile.aim,
  159.         |    tmp7_aim_ip_mobile.rcs_ip
  160.         |    if(tmp4_video. is NULL,0,1) as rcs_video
  161.         |from
  162.         |    tmp7_aim_ip_mobile
  163.         |left join
  164.         |    tmp4_video
  165.         |on
  166.         |    tmp7_aim_ip_mobile.mobile=tmp4_video.mobile
  167.         |""".stripMargin).cache().createOrReplaceTempView("tmp8_aim_ip_video_mobile")
  169.     //SDK数据与号码库拼库
  171.     spark.sql(
  172.       """
  173.         |insert overwrite into cl_nlp.nlp_contact_label_system
  174.         |select
  175.         |    tmp8_aim_ip_mobile.mobile
  176.         |    tmp5_sdk.telcom,
  177.         |    tmp5_sdk.base_province,
  178.         |    tmp5_sdk.base_city,
  179.         |    tmp5_sdk.appplatform,
  180.         |    tmp5_sdk.devicename,
  181.         |    tmp5_sdk.imei,
  182.         |    tmp5_sdk.oaid,
  183.         |    tmp8_aim_ip_mobile.aim,
  184.         |    tmp8_aim_ip_mobile.device,
  185.         |    tmp8_aim_ip_mobile.rcs_ip.
  186.         |    tmp8_aim_ip_mobile.rcs_video
  187.         |from
  188.         |    tmp8_aim_ip_video_mobile
  189.         |left join
  190.         |    tmp5_sdk
  191.         |on
  192.         |    tmp8_aim_ip_video_mobile.mobile = tmp5_sdk.mobileimei
  193.         |""".stripMargin)
  196.     spark.table("tmp1_mobile").unpersist()
  197.     spark.table("tmp2_aim").unpersist()
  198.     spark.table("tmp3_ip").unpersist()
  199.     spark.table("tmp4_video").unpersist()
  200.     spark.table("tmp5_sdk").unpersist()
  201.     spark.table("tmp6_aim_mobile").unpersist()
  202.     spark.table("tmp7_aim_ip_mobile").unpersist()
  203.     spark.table("tmp8_aim_ip_video_mobile").unpersist()
  204.     spark.stop()
  206.   }
  208. }

注：该代码暂未在生产环境中实践，由于数据量巨大这样拼接方式可能耗费较大资源，后续会继续优化

�