标签查询服务是什么？

表4-1展示了用户常住省标签的Hive表结构及数据示例，其中p_date表示标签的数据日期，user_id代表的用户实体ID，province代表用户的常住省。大部分标签与省份标签一样存储在Hive表中，其属性包括用户实体ID、标签信息以及标签时间信息。

表4-1 用户常住省Hive表（userprofile_demo.userprofile_label_province）数据示例

以该表信息为例，当业务希望查询指定用户指定时间下的标签数值时，最直接的方式是编写如下SQL语句从表中查询出标签值数据。

SELECT
province
FROM
userprofile_demo.userprofile_label_province
WHERE
 p_date = '2022-07-11'
AND user_id = 100

当业务请求量较大且对接口响应时间要求比较严格时，直接通过SQL语句从Hive表查询结果的方式不再适用，可以将Hive表中的数据转存到其他存储引擎中来提高数据的访问效率。

以上就是标签查询服务的基本形态，即给定时间和实体ID来获取标签数值。其中“给定时间”一般默认是标签数据的最新日期，以表4-1中数据为例，当查询用户100的省份信息时，如无特殊说明会返回最新日期2022-07-11的标签数据“河南省”，而非2022-07-10下的标签数据“山东省”。

标签查询服务主要应用在以下业务场景中。

• 单用户画像查询：用户画像查询功能可以通过标签查询服务来实现，给定UserId可以查出该用户的多个标签数值并展示在页面上。许多运营类平台在展示用户信息的地方，可以借助标签查询服务获取更多元的画像标签数据来补充用户信息、丰富用户的展示维度。在客服系统或者审核系统中，可以使用标签查询服务获取用户的风险类标签来辅助客服和审核人员做业务判断。
• 运营活动：在运营活动的关键环节可以通过用户标签值来区分运营策略，实现精细化运营。比如在某次红包活动中，可以根据用户的活跃等级来确定用户的红包大小，当用户的活跃等级标签数值为高活时可以分发金额较大的红包。同理，在客户端上判断是否展示某款游戏的入口，可以通过查询用户的“游戏兴趣标签”数值来确定，当兴趣值超过指定阈值时才可以显示游戏入口。
• 算法工程：用户画像标签也属于算法特征，可直接应用到算法模型训练中。推荐系统中大部分模型都提供在线服务，在模型预测时需要输入用户的各类特征数据，通过调用标签查询服务可以获取标签数据作为特征输入模型获得预测结果。推荐系统架构设计中一般会使用特征池存储推荐工程常用的特征和画像数据，借助标签查询服务也可以补充完善该部分数据。

标签数据如何写入缓存

直接从Hive表中查询标签数据响应时间较长且受资源影响无法支持大量的并发请求，为了支持高并发和快速响应，可以将Hive表中的数据转储到其他技术组件中。可用于存储标签数据并支持快速查询的技术组件有挺多，业界常用的有HBase、Elasticsearch、Redis、MongoDB。

虽然标签查询服务业务逻辑简单，但是标签查询需要覆盖不同类型、不同数据结构的标签，这需要标签缓存支持丰富的数据类型。标签查询服务的使用场景多为高并发场景且需要通过灵活扩容来支持更高的并发需求，这一特点要求标签缓存支持高并发调用和较低的扩缩容成本。标签查询服务涉及的标签数据每日都会有大量的更新以及上下线操作，这需要标签缓存支持大数据快速写入以及灵活的数据删减功能。

HBase是Hadoop生态圈下的NoSQL数据库，因其部署成本较高且数据读取较慢，在高并发场景下性能表现较差。Elasticsearch的优势在于全文检索，将标签灌入Elasticsearch往往会配合人群圈选使用，单独用于标签查询时，其数据写入成本较高且不适合高并发调用。MongoDB可用于大数据的存储和随机访问，其性能表现良好，但是在操作的简便性和内存占用上不如Redis。Redis作为一款比较流行的数据缓存被业界广泛使用，其支持丰富的数据结构，可以快速部署高可用的分布式集群，借助Redis的各类特性可以实现标签的灵活上下线。结合标签查询业务特点，最终选择Redis作为标签查询的缓存方案。

将Hive数据同步到Redis的实现方式有多种，以表4-1所示的用户常住省标签数据为例，下面将分别介绍基于Spark、Flink以及自行读取HDFS文件的实现方式。以下代码都假设常住省标签Hive数据表的存储格式为Parquet。

使用Spark写入数据

借助Spark可以简便地读取指定Hive表数据，并写入到Redis中，其核心代码如下所示。

public static void main() {
// 配置Spark运行参数
SparkSession spark = SparkSession.builder()
.appName("Spark Read Hive Data To Redis").enableHiveSupport().getOrCreate();
// 查询Hive表数据
Dataset<Row> sqlDF = spark.sql("SELECT user_id, province FROM userprofile_demo.userprofile_label_province");
// 数据转换为对象
Dataset<LabelInfo> stringsDS = sqlDF.map(
(MapFunction<Row, LabelInfo>) row -> {
return new LabelInfo(row.getLong(0), row.getString(1));
},
Encoders.bean(LabelInfo.class));
// 遍历数据并写入Redis中
stringsDS.foreach(item -> {
String key = String.format("province:uid:%s", item.key);
redisClient.set(key, item.value);
});
}
@AllArgsConstructor
public static class LabelInfo implements Serializable {
private Long key;
private String value;
// ...
}

使用Flink写入数据

Spark是主流的离线大数据处理引擎，Flink主要用于实时大数据处理。目前Flink在推动流批一体化建设，即可以同时支持流式数据处理和离线批数据处理，这也是大数据处理引擎的发展方向。所以通过Flink也可以读取Hive数据并写入Redis，其核心代码如下所示。

public static void main(String[] args) throws Exception {
try {
String inputFile = "Hive数据表HDFS路径";
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
Job job = Job.getInstance();
HadoopInputFormat<Void, Group> hadoopInputFormat = new HadoopInputFormat<Void, Group>(
new ParquetInputFormat(), Void.class, Group.class, job);
ParquetInputFormat.addInputPath(job, new Path(inputFile));
DataSource<Tuple2<Void, Group>> hiveTableData = env.createInput(hadoopInputFormat);
// 处理读取到的Hive表数据
hiveTableData.flatMap(new RichFlatMapFunction<Tuple2<Void, Group>, Tuple2<Void, Group>>() {
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
}
@Override
public void flatMap(Tuple2<Void, Group> voidGroupTuple2,
Collector<Tuple2<Void, Group>> collector) throws InterruptedException {
Group val = voidGroupTuple2.f1;
try {
// 将数据解析为map格式，如{"user_id": 100, "province": "山东省"}
Map<String, String> tempMap = ToolsUtils.stringToMap(val.toString());
insertIntoCache(tempMap);
} catch (Exception e) {
// 异常处理逻辑
}
}
// 将数据写入Redis
public void insertIntoCache(Map<String, String> rawData) {
String key = String.format("province:uid:%s", rawData.get("user_id"));
redisClient.set(key, rawData.get("province"));
}
@Override
public void close() throws Exception {
super.close();
}
}).writeAsText("输出文件地址", FileSystem.WriteMode.OVERWRITE);
// 执行任务
env.execute("DemoCode");
} catch (Exception e) {
// 异常处理逻辑
}
}

自行读取HDFS文件写入数据

通过Spark和Flink可以简便地将Hive表数据缓存到Redis，其实现原理是在分布式场景下分片读取HDFS文件后解析数据内容并最终写入到Redis中。工程上也可以自行编写代码读取HDFS文件并实现数据解析逻辑，其核心代码如下所示。

// 获取Hadoop配置
public static Configuration getConfiguration() {
Configuration conf = new Configuration();
conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
try {
conf.addResource( "mountTable.xml配置文件路径");
conf.addResource("core-site.xml配置文件路径");
conf.addResource("hdfs-site.xml配置文件路径");
} catch (Exception e) {
// 异常处理逻辑
}
return conf;
}
public static void main(String[] args) {
String filePath = "Hive数据表HDFS路径";
List<String> properties = Lists.newArrayList("user_id", "province");
// 构建ParquetReader
Configuration conf = getConfiguration();
Path file = new Path(filePath);
ParquetReader.Builder<Group> builder = ParquetReader.builder(new GroupReadSupport(), file)
.withConf(conf);
ParquetReader<Group> reader = builder.build();
SimpleGroup group = null;
GroupType groupType = null;
// 读取HDFS文件并逐行解析数据内容
while ((group = (SimpleGroup) reader.read()) != null) {
groupType = group.getType();
Map<String, String> dataMap = Maps.newHashMapWithExpectedSize(properties.size());
for (int i = 0; i < groupType.getFieldCount(); i++) {
String colName = groupType.getFieldName(i);
if (properties.contains(colName)) {
String colValue = group.getValueToString(i, 0);
dataMap.put(colName, colValue);
}
}
// 将数据写入Redis
if (!dataMap.isEmpty()) {
String key = String.format("province:uid:%s", dataMap.get("user_id"));
redisClient.set(key, dataMap.get("province"));
}
}
}

以上介绍了3种读取标签Hive表数据并写入到Redis的实现方式，如图4-2所示，其实现逻辑和流程都非常相似。首先读取Hive数据文件，然后将数据内容解析转换成合理的数据结构，最后写入Redis中。

Redis作为标签查询服务的缓存也有一些缺点。Redis本身是基于内存构建的，大规模部署Redis的成本较高；Redis比较适合做热数据缓存，即存储经常被访问到的数据，如果有大量的冷数据是对资源的一种浪费。标签查询场景比较容易出现大量的冷数据，这是因为标签查询场景下大部分用户的标签数据被访问的概率较小。当明确有大量冷数据的时候，可以考虑采用Redis + HBase双层存储的方式，其中Redis用来存储热数据，HBase用来存储冷数据，当在Redis中查询不到数据时兜底从HBase进行查询。比如可以将月活用户的标签数据写入Redis，非月活数据写入HBase，月活用户基本可以覆盖大部分的业务查询需求。双层存储可以节约大量资源，但是其维护成本也会提升，这需要结合自身业务特点和技术能力来做选择。

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