用户画像的标签是如何生成的

本节将结合实际案例介绍各类画像标签的生产方式。离线标签将分别介绍统计类标签、规则类标签和导入类标签，实时标签和挖掘类标签也会结合示例做简要介绍，本节部分环节给出了核心Hive SQL语句及Java代码示例。

统计标签

统计类标签是通过离线数据统计手段，计算出指定时间范围内满足特定要求的标签值。统计类标签大部分涉及时间属性，比如近一天点赞次数、最近一周平均在线时长、最近一个月发布文章数等，但并不是所有统计类标签数据最终都可以量化，比如距今最近一次登录时间、最近一周是否被举报，其结果分别是日期和布尔值。借助大数据引擎执行统计语句可以生产出统计类标签，下面以最近一周平均在线时长、最近一周是否被举报为例，说明统计类标签的生产方式。

“最近一周平均在线时长”标签用于统计最近一周用户在线时长的平均值。假设当前日期是T，其计算过程分为两步：计算出T-7到T-1日期范围内的在线时长总和；用总和除以时间跨度7。假设用户在线时长明细存储在Hive表userprofile_demo.user_online_data的列online_time中，该列类型是bigint，存储的是当日用户在线时长秒数，数据表通过主键是user_id唯一标识一个用户。该统计类标签生成语句如下所示，其中通过SUM函数计算出了每一个user_id的在线时长总和。SQL语句中的日期范围是写死的，在实际生产环节，日期范围可以通过变量来替代。

SELECTuser_id,sum(online_time) / 7FROMuserprofile_demo.user_online_dataWHEREp_date >= '2022-06-20'AND p_date <= '2022-06-26'GROUP BYuser_id

同理，对于“最近一周是否被举报”标签，假设用户举报行为数据存储在行为明细Hive表userprofile_demo.user_report_detail_data中，其中列reported_user_id记录了被举报用户。当前日期是T，其计算过程只需统计出T-7到T-1日期范围内的用户被举报总数，如果总数大于0，则说明用户最近一周被举报过。其统计语句如下所示，该语句涉及子查询语句，需要先统计查询出每一个用户被举报的详细次数，然后在外层查询中根据被举报次数的多少判断最近一周是否被举报，1代表是0代表否。

SELECTt.reported_user_id,CASE WHEN t.reportedCount > 0 THEN '1' ELSE '0' END AS reported_ornotFROM(SELECTreported_user_id,count(1) AS reportedCountFROMuserprofile_demo.user_report_detail_dataWHEREp_date >= '2022-06-20'AND p_date <= '2022-06-26'GROUP BYreported_user_id) t

上述两个统计类标签生产示例SQL语句只是将标签结果查询出来，为了保存标签内容，可以使用insert overwrite将结果写入到Hive表中。

标签生产任务大部分需要例行化执行，在生产环境中需要将上述SQL语句配置成定时调度任务。支持配置大数据调度任务的方式有很多，可以使用Airflow或者DolphinScheduler等开源调度工具，也可以通过编写工程代码实现。以调度工具DolphinScheduler为例，可以在该工具上可视化地配置工作流，并为该工作流配置例行化任务。图3-6展示了通过DolphinScheduler配置标签生产任务的过程，其关键配置包括任务执行的SQL语句以及任务调度周期，SQL语句中可以配置自定义变量来实现数据统计周期的自动变更。

图3-6 DolphinnScheduler标签生成配置截图

规则标签

规则标签的生成依赖现有标签内容，需要在已有标签数据的基础上进行综合条件判断，最终生成新的标签数据，比如“是否男性高粉”依赖性别和粉丝数标签；"Android高端机”依赖手机操作系统和手机价格标签。

以“是否男性高粉”标签为例，假设性别标签存储在userprofile_demo.user_gender_data数据表gender列中，粉丝数标签存储在userprofile_demo.user_fanscount_data数据表fans_count列中，两个表中主键都是user_id。男性高粉的定义是粉丝数超过10万的男性用户，该标签的生成语句如下所示。

-- 写入标签数据表中 --INSERT OVERWRITE TABLE userprofile_demo.user_is_highfans_male_label PARTITION (p_date = '2022-06-26')SELECTt.user_id,-- 判断用户是否属于高粉的男性用户 --CASE WHEN t.gender = 'M' AND t.fans_count > 100000 THEN 1 ELSE 0 END AS is_high_fans_maleFROM(SELECTt1.user_id,t1.gender,t2.fans_countFROM(-- 获取性别和粉丝数明细数据 --SELECTuser_id,genderFROM`userprofile_demo.user_gender_data`WHEREp_date = '2022-06-26') t1INNER JOIN (SELECTuser_id,fans_countFROM`userprofile_demo.user_fanscount_data`WHEREp_date = '2022-06-26') t2 ON (t1.user_id = t2.user_id)) t

上述语句包含三部分内容，首先通过性别和粉丝数标签数据表之间的连接操作查询到所有用户的性别和粉丝数明细数据，然后根据规则找出其中高粉男性用户，最后将统计结果写入到目标数据表中。与统计类标签类似，规则类标签也可以借助现有的调度工具或者工程代码实现标签定时产出。

导入标签

导入类标签依赖用户上传的数据来构建新的标签，用户导入数据的方式主要分为文件上传、从其他数据源导入（如MySQL，Hive）两种方式。比如A调研问卷中的有效用户可以上传到画像平台并构建一个新的标签“A调研重点关注用户”；在B游戏发版后，数据分析师找到了一批潜在的优质用户作为后续重点运营群体，这些用户可以导入到画像平台并构建一个新的标签“B游戏潜在推广用户”。

通过上传文件创建标签需要用户提供包含UserId的文件，在标签管理功能模块上传文件即可。为了保持不同标签的存储方式以及格式相同，通过文件上传创建的标签最终也需要存储到Hive表中。为了实现文件中的数据同步至Hive表，本节将介绍两种实现方案。

方案一：通过SQL语句写入数据

通过insert语句写入Hive表。当上传的文件中数据量较少时，可以通过拼接insert语句将文件中的数据写入到指定Hive表中。在创建“A调研重点关注用户”标签时，上传的文件中包含了一些UserId，可以将这些用户的“A调研重点关注用户”标签取值设置为1，1代表是，0代表否；最终标签数据要存储到Hive表userprofile_demo.a_activity_special_user_data中，该表主要包含的属性为分区健p_date，user_id和is_special标签列，其核心SQL语句如下所示。

INSERT INTOuserprofile_demo.a_activity_special_user_data PARTITION (p_date = '2022-06-26')VALUES(123, 1),(234, 1),....(1000, 1);

该方案需要遍历用户上传的文件数据并解析其中的UserId，然后借助工程代码自动生成上述SQL语句，通过提交SQL语句到大数据引擎，最终实现了通过用户上传文件生成标签的功能。这种执行方式对于数据量级较小的文件是可行的，但当文件数据量较大时通过拼接SQL语句的方式不再适用，一是SQL语句太长容易运行失败，二是SQL语句执行效率较低，大数据量下等待时间较长。

方案二：通过HDFS文件写入数据

通过直接写入HDFS文件的方式快速落盘到Hive表中，该实现方案主要分为两步。

• 解析用户上传的文件，读取文件内容并在当前机器中写入到Parquet格式的文件中。
• 将上述Parquet文件上传到HDFS中，并加载生成Hive表。

该实现方案执行效率较高，但是涉及直接操作HDFS文件，如果代码异常可能污染线上数据。直接写入HDFS文件的方式主要依赖工程实现，其核心代码如下所示。

public void writeDatToParquetFile(List<Long> userIds) throws Exception {// 构建数据表SchemaStringBuilder stringBuilder = new StringBuilder();String ls = System.getProperty("line.separator");stringBuilder.append("message m {").append(ls).append(" optional int64 user_id;").append(ls).append(" optional int64 is_special;").append(ls).append("}");String rawSchema = stringBuilder.toString();// 本地写数据到Parquet文件中String parquetPath = "Parquet文件路径";Path path = new Path(parquetPath);MessageType schema = MessageTypeParser.parseMessageType(rawSchema);GroupFactory factory = new SimpleGroupFactory(schema);Configuration conf = new Configuration();GroupWriteSupport.setSchema(schema, conf);ParquetWriter<Group> writer = CsvParquetWriter.builder(path).withWriteMode(Mode.OVERWRITE).withConf(conf).withCompressionCodec(CompressionCodecName.SNAPPY).build();try {Iterator<Long> iterator = userIds.iterator();while (iterator.hasNext()) {Long userId = iterator.next();Group group = factory.newGroup().append("user_id", userId).append("is_special", 1);writer.write(group);}} finally {writer.close();}// 上传本地文件到HDFS中String destPath = "HDFS文件地址";UserGroupInformation ugi = UserGroupInformation.createRemoteUser("用户名称");ugi.doAs(new PrivilegedExceptionAction<Void>() {public Void run() throws Exception {FileSystem fs = HdfsFileService.getWriteHdfs();fs.copyFromLocalFile(true, true, new Path(parquetPath), new Path(destPath));return null;}});// Load HDFS文件到Hive表中 String loadHdfsFileToTable = "LOAD DATA INPATH " + destPath + " INTO TABLE userprofile_demo.a_activity_special_user_data  PARTITION (p_date='2022-06-26')";// 通过JDBC执行上述语句hiveJdbcTemplate.update(loadHdfsFileToTable);}

综上所述，图3-7展示了生成导入类标签的两种实现方式。

图3-7 导入类标签的两种生成方式

实时标签

在标签分类和标签存储章节都介绍了实时标签，实时标签可以保证标签数据的实时性，能够反馈标签的最新数值。比如“当日实时分享数量”标签，记录了用户从当天凌晨开始到当前时刻的累计分享次数；“当日是否被举报”标签记录了用户当日是否被举报，当举报事件发生时，用户该标签值可以实时更新为“被举报”。

实时标签的数据源一般都是实时数据流，实时数据流中的数据一般来自客户端上报日志或者服务端业务日志。实时数据可以借助消息队列进行传递，下游通过消费队列中的数据来构建实时标签。业界比较流行的消息队列有Kafka、RocketMQ、RabbitMQ和ActiveMQ等，其优劣势和所适用的场景各不相同，其中Kafka比较适用大规模数据传输，在大数据吞吐量和性能上表现良好。用于消费实时数据的技术有Storm、Spark Streaming、Flink等，近几年Flink在业界使用比较广泛，主要因为其支持简单的编程模型，在高吞吐、低延迟、高性能等方面表现良好。

下面以“当日实时分享数量”为例，介绍采用Flink消费Kafka实时数据流来生产实时标签的过程。假设用户分享行为数据记录在user_share_action_detail这一Kafka Topic中，其数据组织为JSON格式，主要包含的属性包括：分享时间shareTime，分享者userId以及视频photoId，数据示例如下所示。

{

"userId": 100, // 用户ID

"photoId": 200, // 分享的视频ID

"shareTime": 1656406377465 // 分享毫秒时间戳

}

通过Flink实时消费Kafka Topic数据，并将实时标签数据更新到Redis中，其主要流程如图3-8所示。

图3-8 当日实时分享数量标签生产流程图

“当日实时分享数量”标签与日期有关，需要区分出不同日期下的标签数据。可以借助分享时间戳计算当前的日期，根据不同日期构建不同的Redis Key前缀，比如dt:20220626和dt:20220627。Redis Key中除了日期之外还要包含实体ID信息，这样便可以快速定位到某个日期下指定UserId的分享数，比如dt:20220626:uid:100:share:count。分享数量标签值可以通过Redis String数据结构存储。当指定UserId在某日期下分享次数增加时，可以通过Redis的incr函数实现标签值变更。

按照业务需求也可以设置Redis Key过期时间，防止存储资源浪费。实时标签数据可以定期从Redis同步到Hive表中，由于所有Key均带有日期前缀，可以从Redis中定期获取指定日期前缀的数据文件，解析数据文件后写入Hive表即可（写入方式可以参考上传文件导入标签）。落盘Hive表后可以作为实时标签的数据备份，也方便后续进行数据回溯和历史数据查询。

实时数据除了用于构建实时标签之外，还可以记录到行为明细数据表中用于明细数据分析（在本书后续章节中介绍）。以用户分享行为为例，Kafka消息中包含的明细数据经由Flink解析后可以写入大数据存储引擎中，为了实现OLAP功能，可以写入ClickHouse表中；为了备份数据并支持离线统计分析，也可以写入到Hive表中。ClickHouse提供数据写入接口，可以在FLink消费实时数据时直接调用其接口实现数据写入；Hive数据的写入可以借助Hive JDBC执行写入语句或者直接写入HDFS文件并加载到Hive表中。实时明细数据写入过程如图3-9所示。

图3-9 实时明细数据写入大数据引擎

挖掘类标签

挖掘类标签是指借助机器学习算法挖掘出的标签。不同于统计和规则类标签，挖掘类标签无法直接通过简单的统计语句计算获取，需要借助算法模型对标签结果进行预测。比如用户的兴趣爱好标签，需要根据用户过往历史行为挖掘出用户的兴趣爱好及概率值；用户的婚育情况标签也无法直接从现有数据中统计获取到，需要借助用户的历史行为进行挖掘，预测用户是否已婚已育。

图3-10展示了挖掘类标签的生产逻辑，算法模型依赖各类特征数据进行模型训练，给定一批待预测用户之后可以计算出标签预测结果，在该预测结果基础上可以封装产出挖掘类标签。大部分挖掘类标签的生产最终都是一个分类问题，可以通过算法找出概率值最大的标签数值，概率的大小代表用户倾向性大小，比如用户已婚的概率是0.8代表用户大概率是已婚状态，该用户可以划分到已婚用户群体中。

图3-10 挖掘类标签生产逻辑

挖掘类标签的生产过程涉及机器学习的一系列环节，其生产流程较长、人力和资源投入较多，目前挖掘类标签在整个标签体系中的占比不高。但是挖掘类标签可以从历史数据中挖掘出用户潜在的标签信息，其可以拓展标签边界范围；在与业务合作过程中，挖掘类标签可以明确优化目标，能够灵活地适配业务需求来取得更好的业务效果。机器学习的常见流程如图3-11所示，大致划分为数据收集与分析、特征工程、模型训练与评估和模型上线几个环节，下面将结合该流程介绍“是否已婚”标签的挖掘生产过程。

图3-11 机器学习流程图

数据收集与分析：对于“是否已婚”标签，业务需求是找到当前已婚的用户，标签取值为是和否，说明该标签挖掘过程是一个二值分类问题。可以预测用户已婚的概率，根据概率值大小进行婚育情况划分。明确需求后可以先收集样本数据，即找到一批真实已婚用户用于后续模型训练和测试使用。有了样本数据后，开始整理后续模型训练可能用到的特征数据，是否已婚可以反映到用户的各类行为数据上，比如用户的浏览记录中如果包含大量婚姻类内容，已婚概率较大；用户的兴趣偏好中如果包含母婴内容，已婚的概率较大；用户的活跃时间反馈用户可以上网的时间分布，已婚用户在时间分布上可能有一定的特点；用户的年龄段如果是中老年则已婚概率较大。数据收集阶段能够找到的数据越多越好，对于收集到的数据要进行数据分析，比如找出数据的最大最小值，方差中位数等；对于字符串类型的数据，要分析不同取值的占比情况。对于质量较差的数据，在该阶段就要直接过滤掉。

特征工程：对于收集到的数据，当质量和数量都满足要求之后，可以筛选出可用的特征数据，特征的好坏最终对模型预测结果有很大的影响。选择出的特征数据需要经过特征处理，如数据清洗、数据规范化以及数据转换。数据清洗用于去掉脏数据和异常数据，数据规范化是统一数据类型及数据格式，数据转换涉及数据的编码、向量化等过程。特征数据经过上述操作之后，可以分为训练数据和测试数据并最终应用到模型训练中。

模型训练与评估：算法模型使用训练数据进行模型训练，通过测试数据来评估模型效果。当预测结果不满足预期时，通过不断调整模型参数来优化模型直到预测结果符合预期。如何选择算法模型是该阶段的重点，需要从决策树、SVM、随机森林、Logistic回归、神经网络等模型中选择最适合解决当前问题的模型，也可以测试不同的算法模型并最终交叉验证选出结果最好的一个。对于模型的评估也有很多方法，比如混淆矩阵、基尼系数、ROC曲线等，需要采用合适的方法评估模型效果。本节中为了预测“是否已婚”概率，采用的是Logistic回归模型。

模型上线：当模型预测结果达到预期之后，可以将模型部署到线上。算法模型导出后主要包含四类文件：Model文件、标签编码文件、元数据文件以及变量文件。“是否已婚”模型文件部署到线上后便可以用于预测用户的“是否已婚”概率值，当概率值超过指定阈值时可以认定为已婚并最终生成标签数据。当后续模型有升级更新时，需要替换线上相关模型文件。

挖掘类标签依赖的机器学习算法及框架目前已经比较成熟。目前常见的机器学习框架有Spark MLlib & Spark ML、Scikit-learn等，框架提供传统的机器学习方法，包括分类，回归，聚类，异常检测和数据准备。深度学习是机器学习的重要组成部分，挖掘类标签也可使用深度学习技术，现在常见的深度学习框架包含TensorFlow，PyTorch以及PaddlePaddle。本节示例“是否已婚”标签属于分类问题，需要预测用户已婚概率，主要使用Spark ML库进行挖掘，其挖掘流程及其核心代码如下所示。

SparkConf conf = new SparkConf().setAppName("ifMarryOrNotDemo");JavaSparkContext jsc = new JavaSparkContext(conf);SQLContext jsql = new SQLContext(jsc);// 准备训练数据List<LabeledPoint> localTraining = Lists.newArrayList(new LabeledPoint(1.0, Vectors.dense(0.0, 1.1, 0.1)),new LabeledPoint(0.0, Vectors.dense(2.0, 1.0, -1.0)),new LabeledPoint(0.0, Vectors.dense(2.0, 1.3, 1.0)),new LabeledPoint(1.0, Vectors.dense(0.0, 1.2, -0.5)));Dataset<Row> training = jsql.applySchema(jsc.parallelize(localTraining), LabeledPoint.class);// 创建LogisticRegression模型实例LogisticRegression lr = new LogisticRegression();// 设置模型参数lr.setMaxIter(10).setRegParam(0.01);// 训练模型LogisticRegressionModel model1 = lr.fit(training);// 准备测试数据List<LabeledPoint> localTest = Lists.newArrayList(new LabeledPoint(1.0, Vectors.dense(-1.0, 1.5, 1.3)),new LabeledPoint(0.0, Vectors.dense(3.0, 2.0, -0.1)),new LabeledPoint(1.0, Vectors.dense(0.0, 2.2, -1.5)));Dataset<Row> test = jsql.applySchema(jsc.parallelize(localTest), LabeledPoint.class);// 测试模型并查询结果model1.transform(test).registerTempTable("results");Dataset<Row> results = jsql.sql("SELECT features, label, probability, prediction FROM results");

本文节选自《用户画像：平台构建与业务实践》，转载请注明出处。