本帖最后由 levycui 于 2022-1-12 19:31 编辑 问题导读： 4、如何实现标签开发及管理？
一、什么是用户画像？ 用户画像简介
用户画像实现步骤 首先，基础数据收集，电商领域大致分为行为数据、内容偏好数据、交易数据，如浏览量、访问时长、家具偏好、回头率等等。而金融领域又有贷款信息，信用卡，各种征信信息等等。 然后，当我们对用户画像所需要的基础数据收集完毕后，需要对这些资料进行分析和加工，提炼关键要素，构建可视化模型。对收集到的数据进行行为建模，抽象出用户的标签。电商领域可能是把用户的基本属性、购买能力、行为特征、兴趣爱好、心理特征、社交网络大致的标签化，而金融风控领域则是更关注用户的基本信息，风险信息，财务信息等等。 随后，要利用大数据的整体架构对标签化的过程进行开发实现，对数据进行加工，将标签管理化。同时将标签计算的结果进行计算。这个过程中需要依靠Hive，Hbase等大数据技术，为了提高数据的实时性，还要用到Flink，Kafka等实时计算技术。 最后，也是最关键的一步，要将我们的计算结果，数据，接口等等，形成服务。比如，图表展示，可视化展示，
用户画像的实时性
二、用户画像系统架构 挑战（一）——大数据
挑战（二）——实时性
挑战（三）——与数仓的结合 Extract，数据抽取，也就是把数据从数据源读出来。 Transform，数据转换，把原始数据转换成期望的格式和维度。如果用在数据仓库的场景下，Transform也包含数据清洗，清洗掉噪音数据。 Load 数据加载，把处理后的数据加载到目标处，比如数据仓库。 DB 是现有的数据来源(也称各个系统的元数据)，可以为mysql、SQLserver、文件日志等，为数据仓库提供数据来源的一般存在于现有的业务系统之中。 ETL的是 Extract-Transform-Load 的缩写，用来描述将数据从来源迁移到目标的几个过程： ODS(Operational Data Store) 操作性数据，是作为数据库到数据仓库的一种过渡，ODS的数据结构一般与数据来源保持一致，便于减少ETL的工作复杂性，而且ODS的数据周期一般比较短。ODS的数据最终流入DW DW (Data Warehouse)数据仓库，是数据的归宿，这里保持这所有的从ODS到来的数据，并长期保存，而且这些数据不会被修改。 DM(Data Mart) 数据集市,为了特定的应用目的或应用范围，而从数据仓库中独立出来的一部分数据，也可称为部门数据或主题数据。面向应用。
架构设计
三、标签体系 什么是标签？
电商类标签体系
金融类标签体系
标签的维度与类型 统计类标签：这类标签是最为基础也最为常见的标签类型，例如对于某个用户来说，他的性别、年龄、城市、星座、近7日活跃时长、近7日活跃天数、近7日活跃次数等字段可以从用户注册数据、用户访问、消费类数据中统计得出。该类标签构成了用户画像的基础； 规则类标签：该类标签基于用户行为及确定的规则产生。例如对平台上“消费活跃”用户这一口径的定义为近30天交易次数>=2。在实际开发画像的过程中，由于运营人员对业务更为熟悉、而数据人员对数据的结构、分布、特征更为熟悉，因此规则类标签的规则确定由运营人员和数据人员共同协商确定； 机器学习挖掘类标签：该类标签通过数据挖掘产生，应用在对用户的某些属性或某些行为进行预测判断。例如根据一个用户的行为习惯判断该用户是男性还是女性，根据一个用户的消费习惯判断其对某商品的偏好程度。该类标签需要通过算法挖掘产生。 标签分级分类
标签命名
最终形成得标签示例：
标签存储与管理 比如用户维度标签： 四、用户画像大数据环境搭建 Ambari搭建底层大数据环境
前期准备 主机 <div>192.168.12.101 master
</div><div>192.168.12.102 slave1
</div><div>192.168.12.103 slave2</div> 复制代码 软件要求 yum和rpm（RHEL / CentOS / Oracle / Amazon Linux） zypper和php_curl（SLES） apt （Debian / Ubuntu） scp, curl, unzip, tar，wget和 gcc* OpenSSL（v1.01，内部版本16或更高版本） Python（带python-devel *） 复制代码 本地仓库
yum install yum-utils createrepo
[root@master ~]# yum -y install httpd
[root@master ~]# service httpd restart
Redirecting to /bin/systemctl restart httpd.service
[root@master ~]# chkconfig httpd on 复制代码 复制代码 createrepo ./
制作本地源 修改文件里边的源地址
vi ambari.repo
vi hdp.repo
#VERSION_NUMBER=2.7.5.0-72
[ambari-2.7.5.0]
#json.url = <a href="http://public-repo-1.hortonworks.com/HDP/hdp_urlinfo.json" target="_blank">http://public-repo-1.hortonworks.com/HDP/hdp_urlinfo.json</a>
name=ambari Version - ambari-2.7.5.0
baseurl=https://username:password@archive.cloudera.com/p/ambari/centos7/2.x/updates/2.7.5.0
gpgcheck=1
gpgkey=https://username:password@archive.cloudera.com/p/ambari/centos7/2.x/updates/2.7.5.0/RPM-GPG-KEY/RPM-GPG-KEY-Jenkins
enabled=1
priority=1
[root@master ambari]# yum clean all
[root@master ambari]# yum makecache
[root@master ambari]# yum repolist 复制代码 软件准备 http://www.oracle.com/technetwor ... nloads-2133151.html rpm -ivh jdk-8u161-linux-x64.rpm
java -version 复制代码 vi /etc/hosts
192.168.12.101 master
192.168.12.102 slave1
192.168.12.103 slave2
复制代码 vi /etc/sysconfig/network
NETWORKING=yes
HOSTNAME=master（其他的节点也对应修改）
关闭防火墙
[root@master~]#systemctl disable firewalld
[root@master~]#systemctl stop firewalld
ssh免密
ssh-keygen
ssh-copy-id -i ~/.ssh/id_rsa.pub remote-host 复制代码 yum install ambari-server 复制代码 Installing : postgresql-libs-9.2.18-1.el7.x86_64 1/4
Installing : postgresql-9.2.18-1.el7.x86_64 2/4
Installing : postgresql-server-9.2.18-1.el7.x86_64 3/4
Installing : ambari-server-2.7.5.0-124.x86_64 4/4
Verifying : ambari-server-2.7.5.0-124.x86_64 1/4
Verifying : postgresql-9.2.18-1.el7.x86_64 2/4
Verifying : postgresql-server-9.2.18-1.el7.x86_64 3/4
Verifying : postgresql-libs-9.2.18-1.el7.x86_64 4/4
Installed:
ambari-server.x86_64 0:2.7.5.0-72
Dependency Installed:
postgresql.x86_64 0:9.2.18-1.el7
postgresql-libs.x86_64 0:9.2.18-1.el7
postgresql-server.x86_64 0:9.2.18-1.el7
Complete!
复制代码 启动与设置 复制代码 复制代码 wget -i -c <a href="http://dev.mysql.com/get/mysql57-community-release-el7-10.noarch.rpm" target="_blank">http://dev.mysql.com/get/mysql57 ... e-el7-10.noarch.rpm</a>
rpm -ivh mysql57-community-release-el7-10.noarch.rpm
yum -y install mysql-community-server
systemctl enable mysqld
systemctl start mysqld.service
systemctl status mysqld.service
grep "password" /var/log/mysqld.log
mysql -uroot -p
set global validate_password_policy=0;
set global validate_password_length=1;
set global validate_password_special_char_count=0;
set global validate_password_mixed_case_count=0;
set global validate_password_number_count=0;
select @@validate_password_number_count,@@validate_password_mixed_case_count,@@validate_password_number_count,@@validate_password_length;
ALTER USER 'root'@'localhost' IDENTIFIED BY 'password';
grant all privileges on . to 'root'@'%' identified by 'password' with grant option;
flush privileges;
exit
yum -y remove mysql57-community-release-el7-10.noarch 复制代码 /opt/ambari/mysql-connector-java-5.1.48.jar 复制代码 mysql -uroot -p
create database ambari;
use ambari
source /var/lib/ambari-server/resources/Ambari-DDL-MySQL-CREATE.sql 复制代码 CREATE USER 'ambari'@'localhost' IDENTIFIED BY 'bigdata';
CREATE USER 'ambari'@'%' IDENTIFIED BY 'bigdata';
GRANT ALL PRIVILEGES ON ambari.* TO 'ambari'@'localhost';
GRANT ALL PRIVILEGES ON ambari.* TO 'ambari'@'%';
FLUSH PRIVILEGES; 复制代码 [root@localhost download]# ambari-server setup
Using python /usr/bin/python
Setup ambari-server
Checking SELinux...
SELinux status is 'enabled'
SELinux mode is 'permissive'
WARNING: SELinux is set to 'permissive' mode and temporarily disabled.
OK to continue [y/n] (y)? y
Customize user account for ambari-server daemon [y/n] (n)? y
Enter user account for ambari-server daemon (root):
Adjusting ambari-server permissions and ownership...
Checking firewall status...
Checking JDK...
[1] Oracle JDK 1.8 + Java Cryptography Extension (JCE) Policy Files 8
[2] Custom JDK
==============================================================================
Enter choice (1): 2
WARNING: JDK must be installed on all hosts and JAVA_HOME must be valid on all hosts.
WARNING: JCE Policy files are required for configuring Kerberos security. If you plan to use Kerberos,please make sure JCE Unlimited Strength Jurisdiction Policy Files are valid on all hosts.
Path to JAVA_HOME: /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.242.b08-0.el7_7.x86_64/jre
Validating JDK on Ambari Server...done.
Check JDK version for Ambari Server...
JDK version found: 8
Minimum JDK version is 8 for Ambari. Skipping to setup different JDK for Ambari Server.
Checking GPL software agreement...
GPL License for LZO: <a href="https://www.gnu.org/licenses/old-licenses/gpl-2.0.en.html" target="_blank">https://www.gnu.org/licenses/old-licenses/gpl-2.0.en.html</a>
Enable Ambari Server to download and install GPL Licensed LZO packages [y/n] (n)? y
Completing setup...
Configuring database...
Enter advanced database configuration [y/n] (n)? y
Configuring database...
==============================================================================
Choose one of the following options:
[1] - PostgreSQL (Embedded)
[2] - Oracle
[3] - MySQL / MariaDB
[4] - PostgreSQL
[5] - Microsoft SQL Server (Tech Preview)
[6] - SQL Anywhere
[7] - BDB
==============================================================================
Enter choice (1): 3
Hostname (localhost):
Port (3306):
Database name (ambari):
Username (ambari):
Enter Database Password (bigdata):
Configuring ambari database...
Enter full path to custom jdbc driver: /opt/ambari/mysql-connector-java-5.1.48.jar
Copying /opt/ambari/mysql-connector-java-5.1.48.jar to /usr/share/java
Configuring remote database connection properties...
WARNING: Before starting Ambari Server, you must run the following DDL directly from the database shell to create the schema: /var/lib/ambari-server/resources/Ambari-DDL-MySQL-CREATE.sql
Proceed with configuring remote database connection properties [y/n] (y)? y
Extracting system views...
.....
Ambari repo file contains latest json url <a href="http://public-repo-1.hortonworks.com/HDP/hdp_urlinfo.json" target="_blank">http://public-repo-1.hortonworks.com/HDP/hdp_urlinfo.json</a>, updating stacks repoinfos with it...
Adjusting ambari-server permissions and ownership...
Ambari Server 'setup' completed successfully. 复制代码 ambari-server start
ambari-server status
ambari-server stop 复制代码 http://<your.ambari.server>:8080 复制代码 集群安装
实时计算环境搭建 Linux系统上安装flink 复制代码 复制代码 复制代码 bin/flink run examples/batch/WordCount.jar 复制代码 $ bin/flink run examples/batch/WordCount.jar
Starting execution of program
Executing WordCount example with default input data set.
Use --input to specify file input.
Printing result to stdout. Use --output to specify output path.
(a,5)
(action,1)
(after,1)
(against,1)
(all,2)
(and,12)
(arms,1)
(arrows,1)
(awry,1)
(ay,1) 复制代码 Druid集群部署 主节点部署 Coordinator 和 Overlord进程 两个数据节点运行 Historical 和 MiddleManager进程 一个查询节点 部署Broker 和 Router进程 conf/druid/cluster/master 复制代码 16 vCPU 122GB内存 2 * 1.9TB SSD 复制代码 复制代码 复制代码 开始部署 tar -xzf apache-druid-0.17.0-bin.tar.gz
cd apache-druid-0.17.0
复制代码 复制代码 conf/druid/cluster/_common/common.runtime.properties 复制代码 druid.metadata.storage.connector.connectURI
druid.metadata.storage.connector.host 复制代码 -- create a druid database, make sure to use utf8mb4 as encoding
CREATE DATABASE druid DEFAULT CHARACTER SET utf8mb4;
-- create a druid user
CREATE USER 'druid'@'localhost' IDENTIFIED BY 'druid';
-- grant the user all the permissions on the database we just created
GRANT ALL PRIVILEGES ON druid.* TO 'druid'@'localhost'; 复制代码 druid.extensions.loadList=["mysql-metadata-storage"]
druid.metadata.storage.type=mysql
druid.metadata.storage.connector.connectURI=jdbc:mysql://<host>/druid
druid.metadata.storage.connector.user=druid
druid.metadata.storage.connector.password=diurd 复制代码 conf/druid/cluster/_common/common.runtime.properties
druid.extensions.loadList=["druid-hdfs-storage"]
#druid.storage.type=local
#druid.storage.storageDirectory=var/druid/segments
druid.storage.type=hdfs
druid.storage.storageDirectory=/druid/segments
#druid.indexer.logs.type=file
#druid.indexer.logs.directory=var/druid/indexing-logs
druid.indexer.logs.type=hdfs
druid.indexer.logs.directory=/druid/indexing-logs
将Hadoop配置XML（core-site.xml，hdfs-site.xml，yarn-site.xml，mapred-site.xml）放在Druid中
conf/druid/cluster/_common/ 复制代码 conf/druid/cluster/_common/ 复制代码 复制代码 启动集群 derby 1527
zk 2181
Coordinator 8081
Overlord 8090 复制代码 Historical 8083
Middle Manager 8091, 8100–8199
复制代码 复制代码 启动主节点 bin/start-cluster-master-no-zk-server 复制代码 bin/start-cluster-data-server 复制代码 bin/start-cluster-query-server 复制代码 五、标签开发 数据接入
高性能 智能的交易重组和操作合并 使用数据库本地接口访问 并行处理体系 binlog
Flink接入Kafka数据
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.11</artifactId>
<version>1.9.0</version>
</dependency> 复制代码 构建FlinkKafkaConsumer val properties = new Properties()properties.setProperty("bootstrap.servers", "localhost:9092")// only required for Kafka 0.8properties.setProperty("zookeeper.connect", "localhost:2181")properties.setProperty("group.id", "test")stream = env .addSource(new FlinkKafkaConsumer[String]("topic", new SimpleStringSchema(), properties)) .print() 复制代码 时间戳和水印 val env = StreamExecutionEnvironment.getExecutionEnvironment()val myConsumer = new FlinkKafkaConsumer[String](...)myConsumer.setStartFromEarliest() // start from the earliest record possiblemyConsumer.setStartFromLatest() // start from the latest recordmyConsumer.setStartFromTimestamp(...) // start from specified epoch timestamp (milliseconds)myConsumer.setStartFromGroupOffsets() // the default behaviour//指定位置//val specificStartOffsets = new java.util.HashMap[KafkaTopicPartition, java.lang.Long]()//specificStartOffsets.put(new KafkaTopicPartition("myTopic", 0), 23L)//myConsumer.setStartFromSpecificOffsets(specificStartOffsets)val stream = env.addSource(myConsumer)
复制代码 检查点 import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import java.util.Properties;
public class KafkaConsumer {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092"); properties.setProperty("group.id", "test"); //构建FlinkKafkaConsumer FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties); //指定偏移量
myConsumer.setStartFromEarliest();
DataStream<String> stream = env
.addSource(myConsumer);
env.enableCheckpointing(5000);
stream.print();
env.execute("Flink Streaming Java API Skeleton");
} 复制代码 数据仓库基础知识
Flink+Hive 添加依赖 flink-connector-hive_2.11-1.10.0.jar
flink-shaded-hadoop-2-uber-2.7.5-8.0.jar
hive-exec-2.3.4.jar 复制代码 <!-- Flink Dependency -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-hive_2.11</artifactId>
<version>1.10.0</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_2.11</artifactId>
<version>1.10.0</version>
<scope>provided</scope>
</dependency>
<!-- Hive Dependency -->
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>${hive.version}</version>
<scope>provided</scope>
</dependency> 复制代码 package com.flink.sql.hiveintegration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;
public class FlinkHiveIntegration {
public static void main(String[] args) throws Exception {
EnvironmentSettings settings = EnvironmentSettings
.newInstance()
.useBlinkPlanner() // 使用BlinkPlanner
.inBatchMode() // Batch模式，默认为StreamingMode
.build();
//使用StreamingMode
/* EnvironmentSettings settings = EnvironmentSettings
.newInstance()
.useBlinkPlanner() // 使用BlinkPlanner
.inStreamingMode() // StreamingMode
.build();*/
TableEnvironment tableEnv = TableEnvironment.create(settings);
String name = "myhive"; // Catalog名称，定义一个唯一的名称表示
String defaultDatabase = "qfbap_ods"; // 默认数据库名称
String hiveConfDir = "/opt/modules/apache-hive-2.3.4-bin/conf"; // hive-site.xml路径
String version = "2.3.4"; // Hive版本号
HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir, version);
tableEnv.registerCatalog("myhive", hive);
tableEnv.useCatalog("myhive");
// 创建数据库，目前不支持创建hive表
String createDbSql = "CREATE DATABASE IF NOT EXISTS myhive.test123";
tableEnv.sqlUpdate(createDbSql);
}
} 复制代码 Flink+Druid <?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 <a href="https://maven.apache.org/xsd/maven-4.0.0.xsd" target="_blank">https://maven.apache.org/xsd/maven-4.0.0.xsd</a>">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.8.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.flinkdruid</groupId>
<artifactId>FlinkDruid</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>FlinkDruid</name>
<description>Flink Druid Connection</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
<version>1.9.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.12</artifactId>
<version>1.9.0</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project> 复制代码 @SpringBootApplication
public class FlinkDruidApp {
private static String url = "http://localhost:8200/v1/post/wikipedia";
private static RestTemplate template;
private static HttpHeaders headers;
FlinkDruidApp() {
template = new RestTemplate();
headers = new HttpHeaders();
headers.setAccept(Arrays.asList(MediaType.APPLICATION_JSON));
headers.setContentType(MediaType.APPLICATION_JSON);
}
public static void main(String[] args) throws Exception {
SpringApplication.run(FlinkDruidApp.class, args);
// Creating Flink Execution Environment
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//Define data source
DataSet<String> data = env.readTextFile("/wikiticker-2015-09-12-sampled.json");
// Trasformation on the data
data.map(x -> {
return httpsPost(x).toString();
}).print();
}
// http post method to post data in Druid
private static ResponseEntity<String> httpsPost(String json) {
HttpEntity<String> requestEntity =
new HttpEntity<>(json, headers);
ResponseEntity<String> response =
template.exchange("http://localhost:8200/v1/post/wikipedia", HttpMethod.POST, requestEntity,
String.class);
return response;
}
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
复制代码 六、用户画像产品化 1、标签展示
2、标签查询
3、标签管理
4、用户分群
https://mp.weixin.qq.com/s/5CcVVlDc2EUAgVMZNvCapQ 最新经典文章，欢迎关注公众号
/2 
京ICP备2020039040号
简书网举报电话：021-34700000
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
显身卡
