首页 文章 精选 留言 我的

【最佳实践】实时计算Flink在广告行业的实时数仓建设实践

2020-08-06 870

行业背景

  • 行业现状： 

    • 广告仍然是互联网公司的主要变现手段，2019年，中国广告市场总体规模达到8674.28亿元，较2018年增长了8.54%，据统计全球互联网市值前十的公司广告收入占比高达40%，可见其重要性。AI、大数据、智能投放等创新技术的普及应用，不仅创生了一批独角兽营销平台，而且大幅拉低了广告投放门槛，拓宽了广告市场空间。

  • 大数据在其行业中的作用：

    • 大数据技术的应用在改变我们生活及工作的同时，为我们寻找数据背后的客观规律提供了一种有效途径。对潜在消费群体进行深入分析，并进行定制营销基础上的现代广告营销，对数据的规模及精准度有着极高的要求，而大数据的出现无疑为其落地提供了强有力的支撑。

业务场景

类似媒体，新闻类等APP，上面有各种广告位提供给广告主。广告主投放广告，用户点击广告将实时的产生操作日志数据，对这些日志数据进行实时分析，通过每个广告位上不同广告的投放地区、广告ID、设备唯一编码等信息，可以统计点击次数、投放次数等指标，可用于制定更高效的广告投放策略，降低投放成本，提高广告收益。

技术架构

image.png


架构解析：
数据采集：该场景中，APP、Web、Server等服务上会产生大量的广告投放、用户广告点击等操作日志数据，这些日志数据被实时采集至日志服务系统（SLS），作为Flink的数据源。
实时数仓架构：该场景中，整个实时数仓构建，全部通过 Flink完成。Flink读取SLS中的原始日志数据，经过数据清洗、数据处理等操作写出到DataHub，Flink进一步读取DataHub的数据进行实时统计分析，最终输出对应的指标结果到RDS，供业务系统使用。

业务指标

  • 实时数据中间层，对原始日志进行实时数据清洗

    • 获取投放主题及维度打宽
    • 获取点击主题及维度打宽

  • 统计投放指标

    • 某个广告在某个省的当天投放量
    • 某个广告在某个市的当天投放量
    • 某个广告在某个投放终端的当天投放量

  • 统计点击指标

    • 某个广告在某个省的当天点击量
    • 某个广告在某个市的当天点击量
    • 某个广告在某个投放终端的当天点击量
  • 热门广告排行榜    

业务代码

场景一：对原始日志进行实时数据清洗

投放主题

根据业务主题分成投放主题和点击主题，当release_status=1时为投放主题。

输入表

create table ods_release(
  `sid` varchar,           --投放请求ID
  exts varchar,                       --扩展信息
  device_type varchar,     --1 android| 2 ios | 9 其他
  release_status varchar,  --投放状态 1 or 2
  device_num varchar,      --设备唯一编码
  release_session varchar, --投放会话ID
  `date` date              --创建时间
) with (
  type ='sls',
...
);

输出表

create table dw_release_exposure(
  release_session varchar, -- comment '投放会话id'
  release_status varchar,  -- comment '投放状态'
  device_num varchar,      -- comment '设备唯一编码'
  device_type varchar,     -- comment '1 android| 2 ios | 9 其他'
  area_code varchar,       -- comment '地区'
  aid varchar,             -- comment '广告id'
  ct date                  -- comment '创建时间'
)with(
type='datahub',
...
);

业务代码

insert into dw_release_exposure
select
  release_session,
  release_status,
  device_num,
  device_type,
  json_value(exts,'$.area_code'),
  json_value(exts,'$.aid'),
  `date` as ct
from
ods_release
where release_status='1'
;

投放主题关联维度表

投放主题与地区维度表、设备维度表进行聚合，得出宽表

输入表

create table dw_release_exposure(
  release_session varchar, -- comment '投放会话id'
  release_status varchar,  -- comment '投放状态'
  device_num varchar,      -- comment '设备唯一编码'
  device_type varchar,     -- comment '1 android| 2 ios | 9 其他'
  area_code varchar,       -- comment '地区'
  aid varchar,             -- comment '广告id'
  ct date                  -- comment '创建时间'
)with(
type='datahub',
...
);

--dim维度表
--（地区，省市，唯一地区编码,编码和city_id是一一对应的）
create table dim_province(
  area_code varchar,
  province_id bigint,
  province_name varchar,
  city_id bigint,
  city_name varchar,
  region_id bigint,
  region_name varchar,
 PRIMARY KEY (area_code),
 PERIOD FOR SYSTEM_TIME--定义维表的变化周期。
 )with(
    type= 'rds',
...
);

--(用户设备维度表）
create table dim_device(
  device_type varchar comment '1 android| 2 ios | 9 其他',
  device_name varchar comment '设备名字',
 PRIMARY KEY (device_type),
 PERIOD FOR SYSTEM_TIME--定义维表的变化周期。
)with(
type= 'rds',
...
);

输出表

create table dm_release_exposure(
  aid varchar,
  aid_count bigint,
  device_name varchar,
  area_code varchar,
  province_id bigint,
  province_name varchar,
  city_id bigint,
  city_name varchar,
  ct date
)with(
type='datahub',
...
);

业务代码

insert into dm_release_exposure
select
  a.aid,
  count(a.aid) aid_count,
  c.device_name,
  a.area_code,
  b.province_id,
  b.province_name,
  b.city_id,
  b.city_name,
  a.ct
from
dw_release_exposure a
join
dim_province  FOR SYSTEM_TIME AS OF PROCTIME() as b on a.area_code=b.area_code
join
dim_device  FOR SYSTEM_TIME AS OF PROCTIME() as c on a.device_type=c.device_type
group by
a.aid,
a.area_code,
a.ct
;

点击主题

根据业务主题分成投放主题和点击主题，当release_status=2时为点击主题。

输入表

create table ods_release(
  `sid` varchar,           --投放请求ID
  exts varchar,                       --扩展信息
  device_type varchar,     --1 android| 2 ios | 9 其他
  release_status varchar,  --投放状态 1 or 2
  device_num varchar,      --设备唯一编码
  release_session varchar, --投放会话ID
  `date` date              --创建时间
  ) with (
  type ='sls',
...
);

输出表

create table dw_release_click(
  release_session varchar,  -- comment '投放会话id'
  release_status varchar,   -- comment '投放状态'
  device_num varchar,       -- comment '设备唯一编码' 
  device_type varchar,      -- comment '1 android| 2 ios | 9 其他'
  `user_id` varchar,          -- comment '用户id'
  area_code varchar,        -- comment '地区'
  aid varchar,              -- comment '广告id'
  ct date                   -- comment '创建时间'
)with(
type='datahub',
...
);

业务代码

insert into dw_release_click
select
  release_session,
  release_status,
  device_num,
  device_type,
  json_value(exts,'$.user_id') as `user_id`,
  json_value(exts,'$.area_code') as area_code,
  json_value(exts,'$.aid') as aid,
  `date` as ct
from
ods_release
where release_status='2'
;

点击主题关联维度表

点击主题与地区维度表进行聚合，得出宽表

输入表

create table dw_release_click(
  release_session varchar,  -- comment '投放会话id'
  release_status varchar,   -- comment '投放状态'
  device_num varchar,       -- comment '设备唯一编码' 
  device_type varchar,      -- comment '1 android| 2 ios | 9 其他'
  area_code varchar,        -- comment '地区'
  aid varchar,              -- comment '广告id'
  user_id varchar,          -- comment '用户id'
  ct date                   -- comment '创建时间'
)with(
type='datahub',
...
);

--dim维度表
--（地区，省市，唯一地区编码,编码和city_id是一一对应的）
create table dim_province(
  area_code varchar,
  province_id bigint,
  province_name varchar,
  city_id bigint,
  city_name varchar,
  region_id bigint,
  region_name varchar,
 PRIMARY KEY (area_code),
 PERIOD FOR SYSTEM_TIME--定义维表的变化周期。
 )with(
    type= 'rds',
...
);

--(用户设备维度表）
create table dim_device(
device_type varchar comment '1 android| 2 ios | 9 其他',
device_name varchar comment '设备名字',
 PRIMARY KEY (device_type),
 PERIOD FOR SYSTEM_TIME--定义维表的变化周期。
)with(
type= 'rds',
...
);

输出表

create table dm_release_click(
  aid varchar,
  aid_count bigint,
  device_name varchar,
  area_code varchar,
  province_id bigint,
  province_name varchar,
  city_id bigint,
  city_name varchar,
  ct date
)with(
type='datahub',
...
);

业务代码

insert into dm_release_click
select
  a.aid,
  count(a.aid) aid_count,
  c.device_name,
  a.area_code,
  b.province_id,
  b.province_name,
  b.city_id,
  b.city_name,
  a.ct
from
dw_release_click a
join
dim_province  FOR SYSTEM_TIME AS OF PROCTIME() as b
on a.area_code=b.area_code
join
dim_device  FOR SYSTEM_TIME AS OF PROCTIME() as c on
a.device_type=c.device_type
group by
a.aid,
a.area_code,
a.ct
;

场景二：统计投放指标

某个广告在某个省的当天投放量

以aid和province_name分组，统计某个广告在某个省的当天投放量

输入表

create table dm_release_exposure(
  aid varchar,
  aid_count bigint,
  device_name varchar,
  area_code varchar,
  province_id bigint,
  province_name varchar,
  city_id bigint,
  city_name varchar,
  ct date
)with(
type='datahub',
...
);

输出表

--某个广告在某个省的当天投放量
CREATE TABLE ads_release_exposure_pro (
    aid                       VARCHAR,
    aid_count                 BIGINT,
    province_name             VARCHAR,
  ct                        DATE,
    primary key(aid,province_name,ct)
) WITH (
    type= 'rds',
...
);

业务代码

insert into ads_release_exposure_pro
select 
  aid,
  sum(aid_count) as aid_count,
  province_name,
  ct
from
dm_release_exposure
group by
aid,
province_name,
ct
;

某个广告在某个市的当天投放量

以aid和city_name分组，统计某个广告在某个市的当天投放量

输入表

create table dm_release_exposure(
  aid varchar,
  aid_count bigint,
  device_name varchar,
  area_code varchar,
  province_id bigint,
  province_name varchar,
  city_id bigint,
  city_name varchar,
  ct date
)with(
type='datahub',
...
);

输出表

CREATE TABLE ads_release_exposure_city (
    aid                   VARCHAR,
    aid_count             BIGINT,
    city_name             VARCHAR,
  ct                    DATE,
    primary key(aid,city_name,ct)
) WITH (
    type= 'rds',
...
);

业务代码

insert into ads_release_exposure_city
select 
  aid,
  sum(aid_count) as aid_count,
  city_name,
  ct
from
dm_release_exposure
group by
aid,
city_name,
ct
;

某个广告在某个投放终端的当天投放量

以aid和device_name分组，统计某个广告在某个用户客户端上的当天投放量

输入表

create table dm_release_exposure(
  aid varchar,
  aid_count bigint,
  device_name varchar,
  area_code varchar,
  province_id bigint,
  province_name varchar,
  city_id bigint,
  city_name varchar,
  ct date
)with(
type='datahub',
...
);

输出表

CREATE TABLE ads_release_exposure_device (
    aid                     VARCHAR,
    aid_count               BIGINT,
    device_name             VARCHAR,
  ct                      DATE,
    primary key(aid,device_name,ct)
) WITH (
    type= 'rds',
...
);

业务代码

insert into ads_release_exposure_device
select
  aid,
  sum(aid_count),
  device_name,
  ct
from
dm_release_exposure
group by 
aid,
device_name,
ct
;

场景三：统计点击指标

某个广告在某个省的当天点击量

以ct和aid、provice_name分组，统计某个广告在某个省的当天点击量

输入表

create table dm_release_click(
  aid varchar,
  aid_count bigint,
  device_name varchar,
  area_code varchar,
  province_id bigint,
  province_name varchar,
  city_id bigint,
  city_name varchar,
  ct date
)with(
type='datahub',
...
);

输出表

CREATE TABLE ads_release_click_pro (
  aid                  VARCHAR,
  aid_count            BIGINT,
  province_name        VARCHAR,
  ct                   DATE,
  primary key(aid,province_name,ct)
) WITH (
  type= 'rds',
...
);

业务代码

insert into ads_release_click_pro
select
  aid,
  count(aid) as aid_count,
  province_name,
  ct
from
dm_release_click
group by
aid,
province_name,
ct
;

某个广告在某个市的当天点击量

以ct和aid、city_name分组，统计某个广告在某个市的当天点击量

输入表

create table dm_release_click(
aid varchar,
aid_count bigint,
device_name varchar,
area_code varchar,
province_id bigint,
province_name varchar,
city_id bigint,
city_name varchar,
ct date
)with(
type='datahub',
...
);

输出表

CREATE TABLE ads_release_click_city (
  aid                  VARCHAR,
  aid_count            BIGINT,
  city_name            VARCHAR,
  ct                   DATE,
  primary key(aid,city_name,ct)
) WITH (
  type= 'rds',
...
);

业务代码

insert into ads_release_click_city
select
aid,
count(aid) as aid_count,
city_name,
ct
from
dm_release_click
group by
aid,
city_name,
ct
;

某个广告在某个投放终端的当天投放量

以aid和device_name分组，统计某个广告在某个用户客户端上的当天投放量

输入表

create table dm_release_click(
  aid varchar,
  aid_count bigint,
  device_name varchar,
  area_code varchar,
  province_id bigint,
  province_name varchar,
  city_id bigint,
  city_name varchar,
  ct date
)with(
type='datahub',
...
);

输出表

CREATE TABLE ads_release_click_device (
  aid                     VARCHAR,
  aid_count               BIGINT,
  device_name             VARCHAR,
  ct                      DATE,
    primary key(aid,device_name,ct)
) WITH (
  type= 'rds',
...
);

业务代码

insert into ads_release_click_device
select
  aid,
  sum(aid_count),
  device_name,
  ct
from
dm_release_exposure
group by
aid,
device_name,
ct
;


场景四：热门广告排行榜

以ct和aid分组，计算当天每个广告的总点击量，对广告ID进行topn排序，得到点击次数最多的三个广告作为最热门广告。根据按天维度的时间字段（ct）和广告ID（aid）分组，计算每天每个广告的总点击量，根据广告ID对点击量进行topn排序，统计得到每天点击次数最多的三个广告，用于数据大屏中的热门广告排行榜。

输入表

create table dm_release_click(
aid varchar,
aid_count bigint,
area_code varchar,
province_id bigint,
province_name varchar,
city_id bigint,
city_name varchar,
ct date
)with(
type='datahub',
...
);

输出表

CREATE TABLE ads_release_click_dtclick (
  Ranking              BIGINT,
    aid                  VARCHAR,
    ct                   DATE,
  aid_count            BIGINT,
  primary key(aid,ct)
) WITH (
    type= 'rds',
...
);

业务代码

INSERT INTO ads_release_click_dtclick
SELECT 
Ranking,
aid,
ct,
aid_count
FROM (
  SELECT *,
     ROW_NUMBER() OVER (PARTITION BY `ct` ORDER BY aid_count desc) AS Ranking
  FROM (
        SELECT 
       `ct` AS `ct`,
        COUNT(aid) AS aid_count,
        aid
        FROM  dm_release_click
        GROUP BY `ct`,aid
    )a
) 
WHERE Ranking <= 3

实时计算 Flink 版产品交流群

test

阿里云实时计算Flink - 解决方案：
https://developer.aliyun.com/article/765097
阿里云实时计算Flink - 场景案例：
https://ververica.cn/corporate-practice
阿里云实时计算Flink - 产品详情页：
https://www.aliyun.com/product/bigdata/product/sc

上一篇 下一篇
优秀的个人博客,低调大师

微信关注我们

原文链接：https://yq.aliyun.com/articles/769797

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

Core Image：iOS图像处理技术追踪

Core Image是苹果官方提供的图像处理框架，通过丰富的built-in（内置）或自定义Filter（过滤器）高效处理静态图片、动态图片或视频。开发者还可以通过构造Filter链或自定义Core Image Kernel来实现更丰富的效果。 在WWDC20中，苹果官方针对Core Image技术在以下三方面做了优化：Core Image对视频/动图的支持、基于Metal构建Core Image (CI) Kernel以及Core Image的Debug支持。 这三方面会在下文逐一提到，文末笔者也会浅谈Core Image在手淘图片库中的应用可能以及对Core Image技术的展望。 优化Core Image对视频/动图的支持 ▐ 创建CIContext 创建 CIContext 时，需要遵循一个 view 一个 context 的原则。由于视频的每一帧都会发生变化，将CIContext的cacheIntermediates属性设置为false可以大大减少内存消耗。 如果在使用Core Image时将同时运用Metal（作为输入或输出），通过设置MTLCommandQueue属性创建...
2020-07-31
696
上一篇

【最佳实践】实时计算Flink在游戏行业的实时数仓建设实践

行业背景 行业现状： 随着互联网和移动互联网的相互促进与融合，以及PC终端和各类移动终端在智能化和便携性上的趋同，游戏产品跨平台运行于各类终端的需求逐步显现，特别是互联网页面游戏中的社交类游戏等产品跨平台运行于各类移动终端已经出现，随着版权价值意识的增强，游戏开发商和运营商在取得版权后，加强了对文化内容的开发利用，力图以多形式多媒介的产品实现版权价值的最大化。 大数据在游戏行业中的作用： 根据游戏数据分析游戏产品趋势，实现精准营销 根据玩家付费和活跃度等进行玩家画像，针对不同的玩家设计不同的商业化活动方案，提升付费玩家的体验，提升游戏消费额 业务场景 某游戏公司开发了个游戏APP，该公司在APP中会发布一些游戏场景、游戏角色、装备、精美皮肤等内容，玩家在线娱乐，产生充值购买等行为。业务的构建涉及到几个端： APP：应用程序，玩家访问入口，玩家主要进行如下操作： 注册账号 在线娱乐 游戏充值 后台系统：对玩家行为数据进行分析，提供给运营/运维人员，用于辅助公司决策。 实时归档日志：用于OLAP查询或离线数据分析 实时KPI统计：统计不同时间段的游戏点击量，作为确定游戏活动开启、版本升级、...
2020-08-07
677
下一篇

相关文章

发表评论

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

时间: 2025-12-12 大小: 211.28KB 下载: 54次
Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

时间: 2025-12-21 大小: 189MB 下载: 2次
Spring

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

时间: 2025-12-17 大小: 5MB 下载: 1次
Rocky Linux

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

时间: 2025-12-17 大小: 1.42GB 下载: 1次
手机查看

扫码在手机上查看文章

扫描二维码，手机阅读更方便

热门标签
Jdk25 Docker OpenAi Rocky SpringCloud Gemini3 HarmonyOS6 Kubernetes Nacos3 Redis DeepSeek SpringBoot4 Service Mesh

欢迎您！

登录后，您将获得更多功能！

热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273