引入
针对用户画像项目来说（产品）必须要支持从多种数据源加载业务数据，构建用户标签。
在之前的标签模型开发中，主要是为了简化开发复杂度，业务数据统一存储到HBase表中。
数据源包含如下几个方面：
存储HDFS文件系统
存储Hive表
存储HBase表
存储MySQL表
存储NoSQL数据库：Redis数据库、MongoDB数据库
存储Elasticsearch索引库
存储Kafka分布式队列
封装数据为RDD：
val offsetRanges = Array(
// topic, partition, inclusive starting offset, exclusive ending offset
OffsetRange(“test”, 0, 0, 100),
OffsetRange(“test”,1, 0, 100),
)val rdd = KafkaUtils.createRDD[String, String](sparkContext, kafkaParams, offsetRanges, PreferConsistent)封装数据为DataFrame：
// Subscribe to multiple topics, specifyingexplicit Kafka offsets
val df = spark
.read
.format(“kafka”)
.option(“kafka.bootstrap.servers”, “host1:port1, host2:port2”)
.option(“subscribe”, “topic1, topic2”)
.option(“startingOffsets”, “””{“topic1”:{“0”:23,”1”:-2},”topic2”:{“0”:-2}}”””)
.option(“endingOffsets”, “””{“topic1”:{“0”:50,”1”:-1},”topic2”:{“0”:-1}}”””)
.load()
df.selectExpr(“CAST(key AS STRING)”, “CAST(value AS STRING)”)
.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
- 、解析规则rule为参数ParamsMap
获取业务标签规则rule，按照分隔符分割数据，具体实现代码如下：
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, SparkSession}/*** 加载业务数据工具类：* 解析业务标签规则rule，依据规则判断数段数据源，加载业务数据*/
object MetaParse extends Logging {/*** 依据标签数据，获取业务标签规则rule，解析转换为Map集合* @param tagDF 标签数据* @return Map集合*/def parseRuleToParams(tagDF: DataFrame): Map[String, String] = {import tagDF.sparkSession.implicits._// 1. 4级标签规则ruleval tagRule: String = tagDF.filter($"level" === 4).head().getAs[String]("rule")logInfo(s"==== 业务标签数据规则: {$tagRule} ====")// 2. 解析标签规则，先按照换行\n符分割，再按照等号=分割/*inType=hbasezkHosts=bigdata-cdh01.itcast.cnzkPort=2181hbaseTable=tbl_tag_logsfamily=detailselectFieldNames=global_user_id,loc_url,log_timewhereCondition=log_time#day#30*/val paramsMap: Map[String, String] = tagRule.split("\n").map{ line =>val Array(attrName, attrValue) = line.trim.split("=")(attrName, attrValue)}.toMap// 3. 返回集合MapparamsMap}/*** 依据inType判断数据源，封装元数据Meta，加载业务数据* @param spark SparkSession实例对象* @param paramsMap 业务数据源参数集合* @return*/def parseMetaToData(spark: SparkSession,paramsMap: Map[String, String]): DataFrame = {// 1. 从inType获取数据源val inType: String = paramsMap("inType")// 2. 判断数据源，封装Meta，获取业务数据val businessDF: DataFrame = inType.toLowerCase match {case "hbase" =>// 解析map集合，封装Meta实体类中val hbaseMeta = HBaseMeta.getHBaseMeta(paramsMap)// 加载业务数据spark.read.format("hbase").option("zkHosts", hbaseMeta.zkHosts).option("zkPort", hbaseMeta.zkPort).option("hbaseTable", hbaseMeta.hbaseTable).option("family", hbaseMeta.family).option("selectFields", hbaseMeta.selectFieldNames).option("filterConditions", hbaseMeta.filterConditions).load()case "mysql" =>// 解析Map集合，封装MySQLMeta对象中val mysqlMeta = MySQLMeta.getMySQLMeta(paramsMap)// 从MySQL表加载业务数据spark.read.format("jdbc").option("driver", mysqlMeta.driver).option("url", mysqlMeta.url).option("user", mysqlMeta.user).option("password", mysqlMeta.password).option("dbtable", mysqlMeta.sql).load()case "hive" =>// Map集合，封装HiveMeta对象val hiveMeta: HiveMeta = HiveMeta.getHiveMeta(paramsMap)// 从Hive表加载数据, TODO：此时注意，如果标签模型业务数从Hive表加载，创建SparkSession对象时，集成Hivespark.read.table(hiveMeta.hiveTable)// def select(cols: Column*): DataFrame, selectFieldNames: _* -> 将数组转换可变参数传递.select(hiveMeta.selectFieldNames: _*)//.filter(hiveMeta.whereCondition)case "hdfs" =>// 解析Map集合，封装HdfsMeta对象中val hdfsMeta: HdfsMeta = HdfsMeta.getHdfsMeta(paramsMap)// 从HDFS加载CSV格式数据spark.read.option("sep", hdfsMeta.sperator).option("header", "true").option("inferSchema", "true").csv(hdfsMeta.inPath).select(hdfsMeta.selectFieldNames: _*)case "es" =>nullcase _ =>// 如果未获取到数据，直接抛出异常new RuntimeException("业务标签规则未提供数据源信息，获取不到业务数据，无法计算标签")null}// 3. 返回加载业务数据businessDF}
}（叠甲：大部分资料来源于黑马程序员，这里只是做一些自己的认识、思路和理解，主要是为了分享经验，如果大家有不理解的部分可以私信我，也可以移步【黑马程序员_大数据实战之用户画像企业级项目】https://www.bilibili.com/video/BV1Mp4y1x7y7?p=201&vd_source=07930632bf702f026b5f12259522cb42，以上，大佬勿喷）
)
