• 用户画像知识点补充——多数据源

    引入

    针对用户画像项目来说（产品）必须要支持从多种数据源加载业务数据，构建用户标签。

    在之前的标签模型开发中，主要是为了简化开发复杂度，业务数据统一存储到HBase表中。

            数据源包含如下几个方面：

            存储HDFS文件系统

            存储Hive表

            存储HBase表

            存储MySQL表

            存储NoSQL数据库：Redis数据库、MongoDB数据库

            存储Elasticsearch索引库

            存储Kafka分布式队列

            封装数据为RDD：

    1. val offsetRanges = Array(
    2. // topic, partition, inclusive starting offset, exclusive ending offset
    3. OffsetRange(“test”, 0, 0, 100),
    4. OffsetRange(“test”,1, 0, 100),
    5. )
    6.  
    7. val rdd = KafkaUtils.createRDD[String, String](sparkContext, kafkaParams, offsetRanges, PreferConsistent)

            封装数据为DataFrame：

    1. // Subscribe to multiple topics, specifyingexplicit Kafka offsets
    2. val df = spark
    3. .read
    4. .format(“kafka”)
    5. .option(“kafka.bootstrap.servers”, “host1:port1, host2:port2”)
    6. .option(“subscribe”, “topic1, topic2”)
    7. .option(“startingOffsets”, “””{“topic1”:{“0”:23,”1”:-2},”topic2”:{“0”:-2}}”””)
    8. .option(“endingOffsets”, “””{“topic1”:{“0”:50,”1”:-1},”topic2”:{“0”:-1}}”””)
    9. .load()
    10. df.selectExpr(“CAST(key AS STRING)”, “CAST(value AS STRING)”)
    11. .as[(String, String)]

    用户数据多种数据源

    进入正题，先说明一下我们的用户画像：

    1)、面向业务的用户标签及用户画像管理中台

    统一规范的标签可视化管理中台，业务人员可自助生产和维护标签，适应营销策略变化。

    2)全端采集用户行为数据，整合业务数据等多种数据源，帮助企业构建体系化用户标签图书馆，输出用户画像，赋能业务实现用户精细化运营和精准营销。

    具体功能说明

    1)、构建用户价值体系

    实现用户召回等精准营销目标

    用户在平台消费几次后，一段时间内没有再次访问平台进行消费，我们需要对其进行流失召回。

    2)、输出全景用户画像

    管理客户全生命周期、有效提升用户体验

    用户群体的特征属性和偏好概况，使用产品的方式是否和预期一致;为精准营销做有效的数据支撑

    3)、利用用户标签形成用户分层

    提供个性化推荐内容，持续提升用户转化

    用户在平台的消费金额达到一定的级别时，根据其消费能力不同，投其所好的推荐不同价格定位的商品。

    4)、利用智能算法，快速找到相似似人群

    补充标签定制的不足，高效锁走目标人群

    选定种子人群，根据特征在更大范围内为每个用户计算相似度，精准找到相似人群，挖掘更多潜在客户

    数据源概述

    针对用户画像标签系统来说，不同标签（业务标签，4级标签）来源于不同的业务数据（订单相关数据、搜搜数据、广告点击数据等）、用户行为数据以及第三方数据（社交数据、信用数据等）构建而来。

    不同类型数据采集存储在不同的存储引擎系统（比如HDFS、HBase、Hive、Elasticsearch、MYSQL数据库等），因此需要用户画像标签系统可以支持从不同的数据源读取业务数据，进行构建标签，恰好Spark SQL支持多数据源的加载与保存。

     

    加载HBase表

    前面的标签开发中，无论是加载注册会员信息表tbl_tag_users还是订单数据表tbl_tag_orders，都是从HBase数据库中读取，自己依据Spark SQL实现外部数据源接口，在标签管理平台构建标签时，通过标签规则rule传递参数，开发标签模型时：解析标签规则获取业务数据，逻辑如下：

     

    重构代码（加载数据）

    将上述代码抽象为两个方法：

    其一：解析标签规则rule为Map集合

    其二：依据规则Map集合中inType判断具体数据源，加载业务数据

    编写MetaParse对象object，创建方法parseRuleToMap和parseMetaToData

    1. 、解析规则rule为参数ParamsMap

    获取业务标签规则rule，按照分隔符分割数据，具体实现代码如下：

    1. import org.apache.spark.internal.Logging
    2. import org.apache.spark.sql.{DataFrame, SparkSession}
    3.  
    4. /**
    5.  * 加载业务数据工具类：
    6.  *      解析业务标签规则rule，依据规则判断数段数据源，加载业务数据
    7.  */
    8. object MetaParse extends Logging {
    9. 	
    10. 	/**
    11. 	 * 依据标签数据，获取业务标签规则rule，解析转换为Map集合
    12. 	 * @param tagDF 标签数据
    13. 	 * @return Map集合
    14. 	 */
    15. 	def parseRuleToParams(tagDF: DataFrame): Map[String, String] = {
    16. 		import tagDF.sparkSession.implicits._
    17. 		
    18. 		// 1. 4级标签规则rule
    19. 		val tagRule: String = tagDF
    20. 			.filter($"level" === 4)
    21. 			.head()
    22. 			.getAs[String]("rule")
    23. 		logInfo(s"==== 业务标签数据规则: {$tagRule} ====")
    24. 		
    25. 		// 2. 解析标签规则，先按照换行\n符分割，再按照等号=分割
    26. 		/*
    27. 			inType=hbase
    28. 			zkHosts=bigdata-cdh01.itcast.cn
    29. 			zkPort=2181
    30. 			hbaseTable=tbl_tag_logs
    31. 			family=detail
    32. 			selectFieldNames=global_user_id,loc_url,log_time
    33. 			whereCondition=log_time#day#30
    34. 		 */
    35. 		val paramsMap: Map[String, String] = tagRule
    36. 			.split("\n")
    37. 			.map{ line =>
    38. 				val Array(attrName, attrValue) = line.trim.split("=")
    39. 				(attrName, attrValue)
    40. 			}
    41. 			.toMap
    42. 		
    43. 		// 3. 返回集合Map
    44. 		paramsMap
    45. 	}
    46. 	
    47. 	/**
    48. 	 * 依据inType判断数据源，封装元数据Meta，加载业务数据
    49. 	 * @param spark SparkSession实例对象
    50. 	 * @param paramsMap 业务数据源参数集合
    51. 	 * @return
    52. 	 */
    53. 	def parseMetaToData(spark: SparkSession,
    54. 	                    paramsMap: Map[String, String]): DataFrame = {
    55. 		
    56. 		// 1. 从inType获取数据源
    57. 		val inType: String = paramsMap("inType")
    58. 		
    59. 		// 2. 判断数据源，封装Meta，获取业务数据
    60. 		val businessDF: DataFrame = inType.toLowerCase match {
    61. 			case "hbase" =>
    62. 				// 解析map集合，封装Meta实体类中
    63. 				val hbaseMeta = HBaseMeta.getHBaseMeta(paramsMap)
    64. 				// 加载业务数据
    65. 				spark.read
    66. 					.format("hbase")
    67. 					.option("zkHosts", hbaseMeta.zkHosts)
    68. 					.option("zkPort", hbaseMeta.zkPort)
    69. 					.option("hbaseTable", hbaseMeta.hbaseTable)
    70. 					.option("family", hbaseMeta.family)
    71. 					.option("selectFields", hbaseMeta.selectFieldNames)
    72. 					.option("filterConditions", hbaseMeta.filterConditions)
    73. 					.load()
    74. 			case "mysql" =>
    75.     		    // 解析Map集合，封装MySQLMeta对象中
    76. 				val mysqlMeta = MySQLMeta.getMySQLMeta(paramsMap)
    77. 				// 从MySQL表加载业务数据
    78. 				spark.read
    79. 					.format("jdbc")
    80. 					.option("driver", mysqlMeta.driver)
    81. 					.option("url", mysqlMeta.url)
    82. 					.option("user", mysqlMeta.user)
    83. 					.option("password", mysqlMeta.password)
    84. 					.option("dbtable", mysqlMeta.sql)
    85. 					.load()
    86. 			case "hive" =>
    87. 				// Map集合，封装HiveMeta对象
    88. 				val hiveMeta: HiveMeta = HiveMeta.getHiveMeta(paramsMap)
    89. 				// 从Hive表加载数据, TODO：此时注意，如果标签模型业务数从Hive表加载，创建SparkSession对象时，集成Hive
    90. 				spark.read
    91.     				.table(hiveMeta.hiveTable)
    92. 					// def select(cols: Column*): DataFrame,   selectFieldNames: _* -> 将数组转换可变参数传递
    93.     				.select(hiveMeta.selectFieldNames: _*)
    94.     				//.filter(hiveMeta.whereCondition)
    95. 			case "hdfs" =>
    96. 				// 解析Map集合，封装HdfsMeta对象中
    97. 				val hdfsMeta: HdfsMeta = HdfsMeta.getHdfsMeta(paramsMap)
    98. 				// 从HDFS加载CSV格式数据
    99. 				spark.read
    100. 					.option("sep", hdfsMeta.sperator)
    101. 					.option("header", "true")
    102. 					.option("inferSchema", "true")
    103.     				.csv(hdfsMeta.inPath)
    104.     				.select(hdfsMeta.selectFieldNames: _*)
    105. 			case "es" =>
    106. 				null
    107. 			case _ =>
    108. 				// 如果未获取到数据，直接抛出异常
    109. 				new RuntimeException("业务标签规则未提供数据源信息，获取不到业务数据，无法计算标签")
    110. 				null
    111. 		}
    112. 		
    113. 		// 3. 返回加载业务数据
    114. 		businessDF
    115. 	}
    116. }

    加载Hive表

    1. import org.apache.spark.sql.Column
    2.  
    3. /**
    4.  * 从Hive表中加载数据，SparkSession创建时与Hive集成已配置
    5. 		inType=hive
    6. 		hiveTable=tags_dat.tbl_logs
    7. 		selectFieldNames=global_user_id,loc_url,log_time
    8. 		## 分区字段及数据范围
    9. 		whereCondition=log_time#day#30
    10.  */
    11. case class HiveMeta(
    12.                    hiveTable: String,
    13.                    selectFieldNames: Array[Column],
    14.                    whereCondition: String
    15.                    )
    16.  
    17. object HiveMeta{
    18. 	
    19. 	/**
    20. 	 * 将Map集合数据解析到HiveMeta中
    21. 	 * @param ruleMap map集合
    22. 	 * @return
    23. 	 */
    24. 	def getHiveMeta(ruleMap: Map[String, String]): HiveMeta = {
    25. 		// 此处省略依据分组字段值构建WHERE CAUSE 语句
    26. 		// val whereCondition = ...
    27. 		
    28. 		// 将选择字段构建为Column对象
    29. 		import org.apache.spark.sql.functions.col
    30. 		val fieldColumns: Array[Column] = ruleMap("selectFieldNames")
    31. 			.split(",")
    32. 			.map{field => col(field)}
    33. 		
    34. 		// 创建HiveMeta对象并返回
    35. 		HiveMeta(
    36. 			ruleMap("hiveTable"), //
    37. 			fieldColumns, //
    38. 			null
    39. 		)
    40. 	}
    41. }

    加载HDFS表

    1. import org.apache.spark.sql.Column
    2.  
    3. /**
    4.  * 从HDFS文件系统读取数据，文件格式为csv类型，首行为列名称
    5. 	inType=hdfs
    6. 	inPath=/apps/datas/tbl_logs
    7. 	sperator=\t
    8. 	selectFieldNames=global_user_id,loc_url,log_time
    9.  */
    10. case class HdfsMeta(
    11. 	                   inPath: String,
    12. 	                   sperator: String,
    13. 	                   selectFieldNames: Array[Column]
    14.                    )
    15.  
    16. object HdfsMeta{
    17. 	
    18. 	/**
    19. 	 * 将Map集合数据解析到HdfsMeta中
    20. 	 * @param ruleMap map集合
    21. 	 * @return
    22. 	 */
    23. 	def getHdfsMeta(ruleMap: Map[String, String]): HdfsMeta = {
    24. 		
    25. 		// 将选择字段构建为Column对象
    26. 		import org.apache.spark.sql.functions.col
    27. 		val fieldColumns: Array[Column] = ruleMap("selectFieldNames")
    28. 			.split(",")
    29. 			.map{field => col(field)}
    30. 		
    31. 		// 创建HdfsMeta对象并返回
    32. 		HdfsMeta(
    33. 			ruleMap("inPath"), //
    34. 			ruleMap("sperator"), //
    35. 			fieldColumns
    36. 		)
    37. 	}
    38. }

    （叠甲：大部分资料来源于黑马程序员，这里只是做一些自己的认识、思路和理解，主要是为了分享经验，如果大家有不理解的部分可以私信我，也可以移步【黑马程序员_大数据实战之用户画像企业级项目】https://www.bilibili.com/video/BV1Mp4y1x7y7?p=201&vd_source=07930632bf702f026b5f12259522cb42，以上，大佬勿喷）

  • 相关阅读:
    mysql经典问题（可用于面试提问）
    Windows 源码编译 nginx （加入nginx-http-flv-module）
    LLM之Prompt（一）：5个Prompt高效方法在文心一言3.5的测试对比
    使用python查找指定文件夹下所有xml文件中带有指定字符的xml文件
    5.13一行代码就能解决的算法题
    Java面试题简答（整理）
    gpt-4o看图说话-根据图片回答问题
    XCTF1-web Robots
    5种在TypeScript中使用的类型保护
    Git查询某次提交属于哪个分支
  • 原文地址：https://blog.csdn.net/qq_41680016/article/details/139380392