引入
针对用户画像项目来说（产品）必须要支持从多种数据源加载业务数据，构建用户标签。
在之前的标签模型开发中，主要是为了简化开发复杂度，业务数据统一存储到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级标签规则rule
val tagRule: String = tagDF
.filter($"level" === 4)
.head()
.getAs[String]("rule")
logInfo(s"==== 业务标签数据规则: {$tagRule} ====")
// 2. 解析标签规则，先按照换行\n符分割，再按照等号=分割
/*
inType=hbase
zkHosts=bigdata-cdh01.itcast.cn
zkPort=2181
hbaseTable=tbl_tag_logs
family=detail
selectFieldNames=global_user_id,loc_url,log_time
whereCondition=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. 返回集合Map
paramsMap
}
/**
* 依据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对象时，集成Hive
spark.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" =>
null
case _ =>
// 如果未获取到数据，直接抛出异常
new RuntimeException("业务标签规则未提供数据源信息，获取不到业务数据，无法计算标签")
null
}
// 3. 返回加载业务数据
businessDF
}
}（叠甲：大部分资料来源于黑马程序员，这里只是做一些自己的认识、思路和理解，主要是为了分享经验，如果大家有不理解的部分可以私信我，也可以移步【黑马程序员_大数据实战之用户画像企业级项目】https://www.bilibili.com/video/BV1Mp4y1x7y7?p=201&vd_source=07930632bf702f026b5f12259522cb42，以上，大佬勿喷）
