AnalyticDB MySQL湖仓版的用户运营分析实践

1. 网络规划

请参考表格中的说明和方案默认示例值为每个规划项做详细规划并在实际部署时将默认示例值修改为您的实际规划。

2. 规划云资源

请参考表格中的说明和方案默认示例值为每个规划项做详细规划并在实际部署时将默认示例值修改为您的实际规划。

1. 创建高权限账号与资源组

在资源部署的阶段，本方案已经创建了云原生数据仓库AnalyticDB MySQL湖仓版集群。接下来，您需要创建一个高权限账号和一个资源组，去执行数据分析的操作。

1. 创建高权限账号：

   1. 在云原生数据仓库AnalyticDB MySQL控制台，在集群列表的湖仓版（3.0）页签，单击刚刚创建的集群。

   2. 在左侧导航栏单击账号管理。然后单击创建账号。

   3. 根据控制台提示，设置账号名称和密码。账号类型选择高权限账号。单击确定。本方案示例的账号为名称testUser，密码为testPassword。请勿在您的环境中使用本方案用于示例的账号名称和密码。

2. 创建资源组并绑定RAM用户：

   1. 在左侧导航栏单击资源管理。然后单击资源组管理。

   2. 单击新增资源组，设置资源组名称（本方案示例为Kafka），任务类型为Job，计算预留资源为0ACU，计算最大资源为8ACU。选择完成后，单击确定。

   3. 单击绑定用户。在用户名下拉框选择RAM用户，然后单击绑定用户。

   4. 在弹出的绑定用户对话框，单击确定。单击右上角关闭图标。

2. 构建测试数据

本阶段构建用于测试的日志数据和维度表数据，为后续数据同步、清洗、分库分表、分析做准备。

1. 构建日志测试数据：

   1. 在消息队列Kafka版控制台实例列表页，单击Kafka实例ID。

   2. 在左侧导航栏，单击Topic 管理。

   3. 在Topic 管理页面，找到刚刚创建的Topic，在操作列单击更多>体验发送消息。

   4. 在消息内容的文本框中，输入JSON格式的消息内容。本方案示例。

      {"appid": "1234321","userid": "1","cost": "200","country": "china","city":"shanghai","appversion": "1.0","clickip": "192.168.0.1","clickTime": "2023-01-01 09:05:01","logintime": "2023-01-01 09:03:00","logouttime": "2023-01-01 10:28:00","reinstalledtime": "2023-01-01 09:00:00","fisrtinstalltime": "2022-01-11 23:10:00","installedtime":"2022-12-11 23:10:00"}
      {"appid": "1234322","userid": "2","cost": "220","country": "china","city":"hangzhou","appversion": "1.3","clickip": "192.168.11.1","clickTime": "2023-01-02 09:25:00","logintime": "2023-01-02 09:30:00","logouttime": "2023-01-02 11:20:00","reinstalledtime": "2023-01-02 09:00:00","fisrtinstalltime": "2022-07-03 23:11:00","installedtime":"2022-12-12 23:11:00"}
      {"appid": "1234322","userid": "3","cost": "40","country": "china","city":"shanghai","appversion": "1.0","clickip": "192.168.0.2","clickTime": "2023-01-03 09:05:01","logintime": "2023-01-03 09:03:00","logouttime": "2023-01-03 10:28:00","reinstalledtime": "2023-01-03 09:00:00","fisrtinstalltime": "2021-12-20 23:10:00","installedtime":"2022-12-13 23:10:00"}
      {"appid": "1234322","userid": "4","cost": "150","country": "china","city":"beijing","appversion": "2.0","clickip": "192.168.11.2","clickTime": "2023-01-04 09:25:00","logintime": "2023-01-04 09:30:00","logouttime": "2023-01-04 11:20:00","reinstalledtime": "2023-01-04 09:00:00","fisrtinstalltime": "2022-12-13 23:11:00","installedtime":"2022-12-13 23:11:00"}

2. 构建维度表测试数据：

   1. 在云数据库RDS控制台示例列表页，单击刚刚创建的RDS实例ID。

   2. 在基本信息页面的右上角，单击登录数据库。

   3. 输入一键配置阶段配置的数据库账号和密码。单击登录。本方案数据库账号默认值为testuser，密码默认值为testPassword01。

   4. 在页面左侧，单击数据库实例，展开数据库实例。

   5. 在已登录实例中，单击之前创建的RDS实例，双击dim数据库，页面出现SQL Console。

   6. 在SQLConsole中输入SQL语句创建维度表。

   CREATE TABLE dim.dim_table(user_id INT PRIMARY KEY,user_name VARCHAR(255),city VARCHAR(255));

   7. 在左上角单击，然后单击全部功能 > 测试数据构建。

   8. 在右上角单击测试数据构建。

   9. 输入任务名称，选择数据库dim和表dim_table。

   10. 在配置算法中，找到user_name列，在生成方式列单击随机（变长字符串 [1, 10]）。在生成方式弹窗中，选择自定义，选择个人信息/英文名字。附加属性选择不限。单击确认算法配置。按相同方法设置city列的生成方式为地址位置/城市。

   11. 设置生成行数，即生成多少行测试数据。本方案以10行为例。

   12. 单击提交申请。

3. 同步日志数据

在构建测试数据的阶段，本方案利用创建的Kafka实例模拟发送了日志数据。接下来您要做的是将Kafka发送的日志数据同步到AnalyticDB MySQL湖仓版。

4. 数据清洗

在数据清洗阶段，我们可以对无用的字段进行删减，并转换部分字段的类型。

5. 分库分表

当前所有的日志数据都在log.clean_data中。接下来我们根据app_id，将不同app的日志数据放在不同的表中。

1. 在Notebook中，添加段落，切换语言为Spark SQL，依次输入SQL创建库、创建两个明细表。

   CREATE DATABASE app_warehouse;

   CREATE TABLE
   app_warehouse.app1_detail (
   app_id BIGINT,
   user_id BIGINT,
   cost DOUBLE,
   click_time STRING,
   firstinstall_time STRING,
   login_time STRING,
   logout_time STRING,
   install_time STRING
   ) USING adb TBLPROPERTIES (
   "distributeType"="HASH",
   "distributeColumns"="install_time",
   "partitionType"="value",
   "partitionCoLumn"="install_time",
   "partitionCount"="30"
   );

   CREATE TABLE
   app_warehouse.app2_detail (
   app_id BIGINT,
   user_id BIGINT,
   cost DOUBLE,
   click_time STRING,
   firstinstall_time STRING,
   login_time STRING,
   logout_time STRING,
   install_time STRING
   ) USING adb TBLPROPERTIES (
   "distributeType"="HASH",
   "distributeColumns"="install_time",
   "partitionType"="value",
   "partitionCoLumn"="install_time",
   "partitionCount"="30"
   );

2. 查询出app1的日志数据并导入app1的日志明细表app_warehouse.app1_detail。如果您希望实现定时自动写入数据到app1_detail表，可以使用DMS任务编排，具体请参见ADB Spark-2。

   INSERT OVERWRITE app_warehouse.app1_detail
   SELECT
   appid,
   userid,
   cost,
   clicktime,
   firstinstalltime,
   logintime,
   logouttime,
   installedtime
   FROM
   log.clean_data
   WHERE
   appid="1234321";

   SELECT * FROM app_warehouse.app1_detail;

3. 按同样方法处理app2的日志数据。如果您希望实现定时自动写入数据到app2_detail表，可以使用DMS任务编排，具体请参见ADB Spark-3。

   INSERT OVERWRITE app_warehouse.app2_detail
   SELECT
   appid,
   userid,
   cost,
   clicktime,
   firstinstalltime,
   logintime,
   logouttime,
   installedtime
   FROM
   log.clean_data
   WHERE
   appid="1234322";

   SELECT * FROM app_warehouse.app2_detail;

6. 同步维度数据

到目前为止，我们已对度量数据进行了清洗、分库分表。接下来我们需要将维度数据同步到AnalyticDB MySQL湖仓版。

1. 登录DMS数据管理服务控制台，在左上角单击，然后单击全部功能 > 集成与开发（DTS） > 数据同步。

2. 单击创建任务。

3. 配置任务名称，配置源库信息和目标库信息。然后单击测试连接以进行下一步。

   配置源库信息

   项目	说明
   选择已有的DMS数据库实例	选择刚刚创建的RDS实例，下方数据库信息将自动填入，您无需重复输入。
   数据库类型	选择MySQL。
   接入方式	选择云实例。
   实例地区	本方案选择华东1（杭州）。
   是否跨阿里云账号	本方案为同一阿里云账号间迁移，选择不跨账号。
   RDS实例ID	自动填充刚刚选择的RDS实例ID。
   数据库账号	自动填充数据库账号。
   数据库密码	自动填充数据库账号的密码。
   连接方式	根据需求选择非加密连接或SSL安全连接。 本方案选择非加密连接。

   配置目标库信息

   选择已有的DMS数据库实例	您可以按实际需求，选择是否使用已有实例。 如使用已有实例，下方数据库信息将自动填入，您无需重复输入。 如不使用已有实例，您需要输入下方的数据库信息。
   数据库类型	选择AnalyticDB MySQL 3.0。
   接入方式	选择云实例。
   实例地区	本方案选择华东1（杭州）。
   实例ID	选择目标云原生数据仓库AnalyticDB MySQL版湖仓版（3.0）集群ID。
   数据库账号	填入目标云原生数据仓库AnalyticDB MySQL版湖仓版（3.0）集群的数据库账号，需具备读写权限。
   数据库密码	填入该数据库账号对应的密码。

4. 在源库对象中选择刚刚创建的dim库，添加到右侧已选择对象中。其他配置保持默认。单击下一步高级配置。

5. 高级配置、配置库表字段、配置库表字段保持默认。预检查通过后，单击下一步购买。

6. 选择付费方式。本方案选择后付费，链路规格为micro。勾选服务条款后，单击购买并启动。

   当运行状态为运行中，表示全量同步已完成，增量同步正常进行中。此时可进行聚合分析。

7. 聚合分析

接下来需要将维度表与分区表进行关联，并进行聚合分析。

1. 登录云原生数据仓库AnalyticDB MySQL控制台，在湖仓版（3.0）集群列表中单击集群ID，在左侧导航栏单击SQL开发。

2. 选择引擎为XIHE，资源组为Interactive。

3. 在SQLConsole窗口，输入SQL。针对不同的应用，分别创建汇总表。

   CREATE TABLE
   app_warehouse.app1_summary (
   app_id bigint,
   active_user_count bigint,
   new_user_count bigint,
   total_income double,
   city string,
   time_in_hour string,
   primary key (time_in_hour)
   ) DISTRIBUTED BY HASH (time_in_hour) ;

   CREATE TABLE
   app_warehouse.app2_summary (
   app_id bigint,
   active_user_count bigint,
   new_user_count bigint,
   total_income double,
   city string,
   time_in_hour string,
   primary key (time_in_hour)
   ) DISTRIBUTED BY HASH (time_in_hour) ;

4. 从明细表中查询数据并导入汇总表。如果您希望实现定时自动写入数据到汇总表，可以在DMS任务编排中实现聚合分析，具体请参见单实例SQL-1和单实例SQL-2。

   INSERT INTO
   app_warehouse.app1_summary
   SELECT
   activeUser.app_id,
   activeUser.active_user_count,
   newUser.new_user_count,
   totalIncome.total_income,
   activeUser.city,
   activeUser.time_in_hour
   FROM
   (SELECT
   a.app_id AS app_id,
   COUNT(DISTINCT a.user_id) AS active_user_count,
   b.city,
   DATE_FORMAT(a.login_time,"%Y%m%d %H:00:00") AS time_in_hour
   FROM
   app_warehouse.app1_detail a
   JOIN dim.dim_table b ON a.user_id=b.user_id
   GROUP by
   time_in_hour,
   b.city
   ) activeUser
   LEFT JOIN (
   SELECT
   COUNT (DISTINCT a.user_id) AS new_user_count,
   b.city,
   DATE_FORMAT(a.firstinstall_time,"%Y%m%d %H:00:00") AS time_in_hour
   FROM
   app_warehouse.app1_detail a
   JOIN dim.dim_table b ON a.user_id=b.user_id
   GROUP BY
   time_in_hour,
   b.city
   ) newUser ON activeUser.time_in_hour=newUser.time_in_hour
   AND activeUser.city=newUser.city
   LEFT JOIN (
   SELECT
   ROUND(SUM(a.cost), 2) AS total_income,
   b.city,
   DATE_FORMAT(a.click_time,"%Y%m%d %H:00:00") AS time_in_hour
   FROM
   app_warehouse.app1_detail a
   JOIN dim.dim_table b ON a.user_id=b.user_id
   GROUP BY
   time_in_hour,
   b.city
   ) totalIncome ON activeUser.time_in_hour=totalIncome.time_in_hour
   and activeUser.city=totalIncome.city;

   INSERT INTO
   app_warehouse.app2_summary
   SELECT
   activeUser.app_id,
   activeUser.active_user_count,
   newUser.new_user_count,
   totalIncome.total_income,
   activeUser.city,
   activeUser.time_in_hour
   FROM
   (SELECT
   a.app_id AS app_id,
   COUNT(DISTINCT a.user_id) AS active_user_count,
   b.city,
   DATE_FORMAT(a.login_time,"%Y%m%d %H:00:00") AS time_in_hour
   FROM
   app_warehouse.app2_detail a
   JOIN dim.dim_table b ON a.user_id=b.user_id
   GROUP by
   time_in_hour,
   b.city
   ) activeUser
   LEFT JOIN (
   SELECT
   COUNT (DISTINCT a.user_id) AS new_user_count,
   b.city,
   DATE_FORMAT(a.firstinstall_time,"%Y%m%d %H:00:00") AS time_in_hour
   FROM
   app_warehouse.app2_detail a
   JOIN dim.dim_table b ON a.user_id=b.user_id
   GROUP BY
   time_in_hour,
   b.city
   ) newUser ON activeUser.time_in_hour=newUser.time_in_hour
   AND activeUser.city=newUser.city
   LEFT JOIN (
   SELECT
   ROUND(SUM(a.cost), 2) AS total_income,
   b.city,
   DATE_FORMAT(a.click_time,"%Y%m%d %H:00:00") AS time_in_hour
   FROM
   app_warehouse.app2_detail a
   JOIN dim.dim_table b ON a.user_id=b.user_id
   GROUP BY
   time_in_hour,
   b.city
   ) totalIncome ON activeUser.time_in_hour=totalIncome.time_in_hour
   and activeUser.city=totalIncome.city;

5. 查询汇总表数据。

   SELECT * FROM app_warehouse.app1_summary;

   SELECT * FROM app_warehouse.app2_summary;

8. 任务调度

您还可以通过DMS任务编排，实现自动任务调度，提高数据开发效率。

1. 在DMS中登录AnalyticDB MySQL。

   1. 登录云原生数据仓库AnalyticDB MySQL控制台，在湖仓版（3.0）集群列表中单击集群ID，然后在右上角单击登录数据库。

   2. 输入数据库账号和数据库密码，单击登录。

   3. 在白名单问题对话框中，单击复制IP网段。

   4. 在云原生数据仓库AnalyticDB MySQL控制台，单击集群管理 > 集群信息。

   5. 在数据安全-白名单设置区域，单击default后面的修改。在组内白名单中粘贴步骤c复制的IP网段，然后单击确定。

   6. 关闭白名单问题对话框。在登录实例对话框中，单击登录。

2. 在左上角单击，然后单击任务编排。

3. 新增任务流。

   1. 单击新增任务流。

   2. 在新建任务流对话框中，输入任务流名称和描述，单击确认。

4. 从左侧任务列表中，拖拽3个ADB Spark节点，2个单实例SQL节点到画布中，并连接成任务流。请根据您的业务场景设计任务流。本方案示例所示。

   

5. 单击任务节点并选择，配置任务节点。

ADB Spark-1

1. 选择地域为华东1（杭州）。

2. 选择ADB实例为刚刚创建的AnalyticDB MySQL湖仓版集群。

3. 选择ADB资源组为刚刚创建的资源组。

4. 任务类型选择SQL。

5. 任务名称可自定义。

6. 在作业配置中输入SQL语句。

-- Here is just an example of SparkSQL. Modify the content and run your spark program.
set spark.driver.resourceSpec=medium;
set spark.executor.instances=2;
set spark.executor.resourceSpec=medium;
-- Here are your sql statements
INSERT INTO
log.clean_data PARTITION (installedtime)
SELECT
appid,
userid,
cast(cost as double),
city,
clicktime,
fisrtinstalltime,
logintime,
logouttime,
installedtime
FROM
kafka.user_log;

7. 单击保存。关闭ADB Spark-1页签。

ADB Spark-2

1. 选择地域为华东1（杭州）。

2. 选择ADB实例为刚刚创建的AnalyticDB MySQL湖仓版集群。

3. 选择ADB资源组为刚刚创建的资源组。

4. 任务类型选择SQL。

5. 任务名称可自定义。

6. 在作业配置中输入SQL语句。

-- Here is just an example of SparkSQL. Modify the content and run your spark program.
set spark.driver.resourceSpec=medium;
set spark.executor.instances=2;
set spark.executor.resourceSpec=medium;
-- Here are your sql statements
INSERT OVERWRITE app_warehouse.app1_detail
SELECT
appid,
userid,
cost,
clicktime,
firstinstalltime,
logintime,
logouttime,
installedtime
FROM
log.clean_data
WHERE
appid="1234321";

7. 单击保存。关闭ADB Spark-2页签。

单实例SQL-1

1. 数据库选择之前创建的app_warehouse。

2. 输入聚合分析的SQL语句。

INSERT INTO
app_warehouse.app1_summary
SELECT
activeUser.app_id,
activeUser.active_user_count,
newUser.new_user_count,
totalIncome.total_income,
activeUser.city,
activeUser.time_in_hour
FROM
(SELECT
a.app_id AS app_id,
COUNT(DISTINCT a.user_id) AS active_user_count,
b.city,
DATE_FORMAT(a.login_time,"%Y%m%d %H:00:00") AS time_in_hour
FROM
app_warehouse.app1_detail a
JOIN dim.dim_table b ON a.user_id=b.user_id
GROUP by
time_in_hour,
b.city
) activeUser
LEFT JOIN (
SELECT
COUNT (DISTINCT a.user_id) AS new_user_count,
b.city,
DATE_FORMAT(a.firstinstall_time,"%Y%m%d %H:00:00") AS time_in_hour
FROM
app_warehouse.app1_detail a
JOIN dim.dim_table b ON a.user_id=b.user_id
GROUP BY
time_in_hour,
b.city
) newUser ON activeUser.time_in_hour=newUser.time_in_hour
AND activeUser.city=newUser.city
LEFT JOIN (
SELECT
ROUND(SUM(a.cost), 2) AS total_income,
b.city,
DATE_FORMAT(a.click_time,"%Y%m%d %H:00:00") AS time_in_hour
FROM
app_warehouse.app1_detail a
JOIN dim.dim_table b ON a.user_id=b.user_id
GROUP BY
time_in_hour,
b.city
) totalIncome ON activeUser.time_in_hour=totalIncome.time_in_hour
and activeUser.city=totalIncome.city;

3. 单击保存。关闭单实例SQL-1页签。

ADB Spark-3

1. 选择ADB实例为刚刚创建的AnalyticDB MySQL湖仓版集群。

2. 选择ADB资源组为刚刚创建的资源组。

3. 任务类型选择SQL。

4. 任务名称可自定义。

5. 在作业配置中输入SQL语句。

-- Here is just an example of SparkSQL. Modify the content and run your spark program.
set spark.driver.resourceSpec=medium;
set spark.executor.instances=2;
set spark.executor.resourceSpec=medium;
-- Here are your sql statements
INSERT OVERWRITE app_warehouse.app2_detail
SELECT
appid,
userid,
cost,
clicktime,
firstinstalltime,
logintime,
logouttime,
installedtime
FROM
log.clean_data
WHERE
appid="1234322";

6. 单击保存。关闭ADB Spark-3页签。

单实例SQL-2

1. 数据库选择之前创建的app_warehouse。

2. 输入聚合分析的SQL语句。

INSERT INTO
app_warehouse.app2_summary
SELECT
activeUser.app_id,
activeUser.active_user_count,
newUser.new_user_count,
totalIncome.total_income,
activeUser.city,
activeUser.time_in_hour
FROM
(SELECT
a.app_id AS app_id,
COUNT(DISTINCT a.user_id) AS active_user_count,
b.city,
DATE_FORMAT(a.login_time,"%Y%m%d %H:00:00") AS time_in_hour
FROM
app_warehouse.app2_detail a
JOIN dim.dim_table b ON a.user_id=b.user_id
GROUP by
time_in_hour,
b.city
) activeUser
LEFT JOIN (
SELECT
COUNT (DISTINCT a.user_id) AS new_user_count,
b.city,
DATE_FORMAT(a.firstinstall_time,"%Y%m%d %H:00:00") AS time_in_hour
FROM
app_warehouse.app2_detail a
JOIN dim.dim_table b ON a.user_id=b.user_id
GROUP BY
time_in_hour,
b.city
) newUser ON activeUser.time_in_hour=newUser.time_in_hour
AND activeUser.city=newUser.city
LEFT JOIN (
SELECT
ROUND(SUM(a.cost), 2) AS total_income,
b.city,
DATE_FORMAT(a.click_time,"%Y%m%d %H:00:00") AS time_in_hour
FROM
app_warehouse.app2_detail a
JOIN dim.dim_table b ON a.user_id=b.user_id
GROUP BY
time_in_hour,
b.city
) totalIncome ON activeUser.time_in_hour=totalIncome.time_in_hour
and activeUser.city=totalIncome.city;

3. 单击保存。关闭单实例SQL-2页签。

6. 单击页面下方的任务流信息，打开开启调度的开关。根据实际业务场景设置调度类型、生效时间、调度周期、具体时间。本方案示例如下：调度类型为定时调度/周期调度，生效时间为2023-08-07至2023-08-15，调度周期为日，具体时间为00:00。

7. 单击发布。

9. 可视化分析报告

接下来您要将数据制作成可视化分析报告。

1. 在Quick BI控制台，根据控制台提示填写试用信息，单击开启专业版试用。

2. 在左侧导航栏单击数据源。

3. 单击Alibaba AnalyticDB for MySQL。

4. 在配置连接页面，填写连接配置。

   项目	说明	示例
   连接方式	本方案选择自动连接。	自动连接
   选择数据库	选择刚刚创建的AnalyticDB MySQL湖仓版（3.0）集群。	amv-bp1tv9mhg1******
   显示名称	自定义数据源的名称。	adb
   数据库地址	填写AnalyticDB MySQL湖仓版（3.0）集群的VPC地址。 如果您无法提供阿里云账号的AccessKeyID和AccessKeySecret，此处请填写公网地址。	amv-bp1tv9mhg1n*******0001031.ads.aliyuncs.com
   端口	3306。	3306
   数据库	填写汇总表所在的数据库。	app_warehouse
   账号	填写AnalyticDB MySQL湖仓版（3.0）集群具有读权限的账号。本方案以刚刚创建的高权限账号为例。	testUser
   密码	填写AnalyticDB MySQL湖仓版（3.0）集群账号的密码。	testPassword
   VPC数据源	勾选后，通过VPC网络访问数据库。	勾选
   购买者AccessId	填写阿里云账号的AccessKeyID。	LTAI5t9YYqutv5svx1******
   购买者AccessKey	填写阿里云账号的AccessKeySecret。	5ZPYdKE4AblC4j4hZhWQycMv******
   实例ID	填写AnalyticDB MySQL湖仓版（3.0）集群ID。	amv-bp1tv9mhg1******
   区域	填写AnalyticDB MySQL湖仓版（3.0）集群所在的地域。	华东1（杭州）

5. 将Quick BI的IP地址添加到AnalyticDB MySQL湖仓版（3.0）集群的白名单。

   1. 在云原生数据仓库AnalyticDB MySQL控制台，在左侧导航栏单击集群列表。

   2. 在湖仓版（3.0）页签中，单击集群名称。

   3. 在集群信息页的数据安全-白名单设置区域，单击创建白名单分组，设置分组名称为quickbi，在组内白名单中粘贴Quick BI的IP地址，单击确定。

6. 单击连接测试。控制台提示数据源连通性正常，然后单击确定。

7. 单击新建数据集。将想要分析的汇总表app2_summary拖拽到工作区，在右上角单击保存。将数据集命名为app2_summary，单击确定。

8. 在右上角单击开始分析 > 创建仪表板。

9. 从右侧的数据区域，拖拽维度和度量到字段区域，生成图表。

   • 拖拽维度time_in_hour和度量active_user_count，展示每小时的活跃用户数。

   • 拖拽维度time_in_hour和度量new_user_count，展示每小时的新用户数。

   • 拖拽维度city和度量total_income，展示每个城市的付费总额。