Flink 实时数仓(二)【ODS 层开发】

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
实时数仓Hologres,5000CU*H 100GB 3个月
云原生数据仓库AnalyticDB MySQL版,8核32GB 100GB 1个月
简介: Flink 实时数仓(二)【ODS 层开发】

前言

       最近投了不少的实习,也收到不错的反馈,虽然是中小公司偏多,但是毕竟现在这个环境双非进大厂实习可不同当年了。可惜的是学院不放人,无奈啊,遍身罗绮者,不是养蚕人。我累死累活肝了两年了,好不容易找到个不错的实习,可是学院...

       不骂了,没那功夫扯淡。回归主题,实时数仓将来真要是有机会从事的话那真不错,也不枉我当时学Flink的时候花了一整个学期,没暖气没电源的教室跑着三台虚拟机,CPU爆满,每天学俩小时就没电了,但还是花了半个学期学完了。今天开始正式的数仓搭建。

1、ODS 层开发

       在实时数仓这里,当我们把数据采集到 Kafka (topic_log 和 topic_db 主题)的时候,其实 ODS 层的数据存储任务就已经完成了(ODS 层的任务:数据的存储和备份)。接下来我们需要做的就是保持数据的有序:

1.1、Kafka 数据有序

       Kafka 只能保证单分区内有序,并不能保证全局有序。

1.1.1、设置 Kafka 分区默认个数

       这里我们需要设置 Kafka 的分区个数为 4,毕竟实时数仓对数据的吞吐量、并发性能的要求是比较高的,所以我们不能为了数据的有序性而把数据到挤到一个分区中:

vim /opt/module/kafka_2.12-3.4.1/config/server.properties
// 修改配置
num.partition=4

1.1.2、设置 Flink 精准一次

       Flink 程序从 Kafka 消费数据时会启动同属于一个消费者组的四个消费者,Kafka 消费者的默认分区分配策略是 Range + CooperativeSticky,消费者数和分区数相同时,每个消费者消费一个分区的数据。只要单分区数据有序,即可保证 Flink 单个并行度数据有序。

       我们这里的 Kafka 版本是 3.0.0,在 Kafka 1.x 及之后的版本中,保证数据单分区有序,条件如下:

不开启 Kafka 幂等性的情况

max.in.flight.requests.per.connection=1

开启 Kafka 幂等性的情况

// 必须小于等于5
max.in.flight.requests.per.connection=5

       在kafka1.x以后,启用幂等后,kafka服务端会缓存producer发来的最近5个request的元数据,故无论如何,都可以保证最近5个request的数据都是有序的。

       默认情况下幂等性是开启的,max.in.flight.requests.per.connection 默认值为 5,所以单分区数据默认是有序的,不需要做任何配置。

综上,我们可以保证 Flink 程序单个并行度的数据有序。

1.2、Maxwell 同步历史维度数据

       我们的实时计算不需要考虑历史的事实数据(比如下单、加购),但要考虑历史维度数据,毕竟没有维度只有业务过程是没法进行计算的。

1.2.1、为什么要同步历史维度数据而不同步历史业务数据

       当一个用户进行下单这个业务过程的时候,比如 user_id 为 1001 的用户在 province_id 为 17 的省份下单了 sku_id 为 10 的商品,并使用了 cupon_id 为 1001 的优惠券。

       当这个数据传到我们实时数仓的时候,我们必须知道用户是谁,它买了什么东西,有没有使用优惠券、使用了什么类型的优惠券、这个省份 id 是什么地方、下单方式是什么。这就涉及到了很多维度数据,所以我们必须提前把维度数据准备好,等到数据来的时候直接拿来用而不是才去业务库同步。

       在 ODS 层这里我们只需要原封不动的把维度数据导入到 Kafka ,等到搭建 DIM 层的时候直接从 ODS 层拿,而不是让 DIM 层去业务数据库中同步。

       在我们这个项目中,我们需要通过 Maxwell 同步下面这些维度表到 Kafka 的

activity_info
activity_rule
activity_sku
base_category1
base_category2
base_category3
base_province
base_region
base_trademark
coupon_info
coupon_range
financial_sku_cost
sku_info
spu_info
user_info

编写同步脚本:

#!/bin/bash
 
# 该脚本的作用是初始化所有的业务数据,只需执行一次
 
MAXWELL_HOME=/opt/module/maxwell
 
import_data() {
    for tab in $@
    do
      $MAXWELL_HOME/bin/maxwell-bootstrap --database gmall --table ${tab} --config $MAXWELL_HOME/config.properties
    done
}
 
case $1 in
activity_info | activity_rule | activity_sku | base_category1 | base_category2 | base_category3 | base_dic | base_province | base_region | base_trademark | coupon_info | coupon_range | financial_sku_cost | sku_info | spu_info | user_info)
  import_data $1 
  ;;
"all")
  import_data activity_info activity_rule activity_sku base_category1 base_category2 base_category3 base_dic base_province base_region base_trademark coupon_info coupon_range financial_sku_cost sku_info spu_info user_info
  ;;
esac

思考

       我们之前在学习离线数仓的时候, 使用 Maxwell 和 DataX 来同步业务数据,其中的 Maxwell 在离线数仓中其实并没有什么作用,至于削峰解耦在离线数仓中是根本不用考虑的。而如果不使用 Kafka ,我们可以直接通过 Flume 直接采集到 HDFS。

       在离线数仓中使用 Maxwell 的作用完全是为了现在学习实时数仓时,方便 Flink 来直接从 Kafka 去读取数据。但是 Flume 的数据中包含的 Event Header ,它对于实时数仓来说是完全没有用的,所以我们当时为了妥协实时数仓,就把 Flume 数据中的 Header 给去掉了,但是也就引入了零点漂移的问题,毕竟 Event Header 中保存着 timestamp 信息,而它在经过 Kafka 之后,会被 Kafka 给它添加一个 Header ,Header 中的 timestamp 时间默认为 Kafka 处理的时间,所以我们当时又设置了 Flume 拦截器来把 Header 中的 timestamp 值设置为 body 中的时间戳(因为拦截器只能设置在 Source 和 Channel 之间,所以还需要一个 Flume 再从 Kafka 读出来)。

Flume 拦截器代码

import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
 
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
 
public class TimeStampInterceptor implements Interceptor {
 
    private ArrayList<Event> events = new ArrayList<>();
 
    @Override
    public void initialize() {
 
    }
 
    @Override
    public Event intercept(Event event) {
 
        Map<String, String> headers = event.getHeaders();
        String log = new String(event.getBody(), StandardCharsets.UTF_8);
 
        JSONObject jsonObject = JSONObject.parseObject(log);
 
        String ts = jsonObject.getString("ts");
        headers.put("timestamp", ts);
 
        return event;
    }
 
    @Override
    public List<Event> intercept(List<Event> list) {
        events.clear();
        for (Event event : list) {
            events.add(intercept(event));
        }
 
        return events;
    }
 
    @Override
    public void close() {
 
    }
 
    public static class Builder implements Interceptor.Builder {
        @Override
        public Interceptor build() {
            return new TimeStampInterceptor();
        }
 
        @Override
        public void configure(Context context) {
        }
    }
}

总结

       实时数仓中 ODS 层的工作很简单,我们只需要用 Maxwell 把业务数据库中的维度表进行实时同步即可。至于服务器里的日志数据我们根本不需要实时处理!一个日志数据没必要实时处理!所以实时数仓中我们也就不需要用 Flume 这个工具。

       所以对于实时数仓的 ODS 层,我们主要用的就是 Maxwell 来同步维度数据,而对于事实数据(比如下单、加购),等到 DWD 层再进行处理。

相关实践学习
AnalyticDB MySQL海量数据秒级分析体验
快速上手AnalyticDB MySQL,玩转SQL开发等功能!本教程介绍如何在AnalyticDB MySQL中,一键加载内置数据集,并基于自动生成的查询脚本,运行复杂查询语句,秒级生成查询结果。
阿里云云原生数据仓库AnalyticDB MySQL版 使用教程
云原生数据仓库AnalyticDB MySQL版是一种支持高并发低延时查询的新一代云原生数据仓库,高度兼容MySQL协议以及SQL:92、SQL:99、SQL:2003标准,可以对海量数据进行即时的多维分析透视和业务探索,快速构建企业云上数据仓库。 了解产品 https://www.aliyun.com/product/ApsaraDB/ads
相关文章
|
1月前
|
存储 数据采集 JavaScript
深入理解数仓开发(一)数据技术篇之日志采集
深入理解数仓开发(一)数据技术篇之日志采集
|
1月前
|
消息中间件 关系型数据库 Kafka
深入理解数仓开发(二)数据技术篇之数据同步
深入理解数仓开发(二)数据技术篇之数据同步
|
16天前
|
存储 SQL 消息中间件
Hologres+Flink企业级实时数仓核心能力介绍
通过Hologres+Flink构建易用、统一的企业级实时数仓。
|
4天前
|
数据采集 运维 Cloud Native
Flink+Paimon在阿里云大数据云原生运维数仓的实践
构建实时云原生运维数仓以提升大数据集群的运维能力,采用 Flink+Paimon 方案,解决资源审计、拓扑及趋势分析需求。
333 0
Flink+Paimon在阿里云大数据云原生运维数仓的实践
|
28天前
|
Java 数据处理 Apache
实时计算 Flink版产品使用问题之lookup Join hologres的维表,是否可以指定查bitmap
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
28天前
|
JSON 关系型数据库 MySQL
实时计算 Flink版产品使用问题之在使用CDAS语法同步MySQL数据到Hologres时,如果开启了字段类型宽容模式,MySQL中的JSON类型会被转换为什么
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
1月前
|
Cloud Native 数据管理 OLAP
云原生数据仓库AnalyticDB产品使用合集之是否可以创建表而不使用分区
阿里云AnalyticDB提供了全面的数据导入、查询分析、数据管理、运维监控等功能,并通过扩展功能支持与AI平台集成、跨地域复制与联邦查询等高级应用场景,为企业构建实时、高效、可扩展的数据仓库解决方案。以下是对AnalyticDB产品使用合集的概述,包括数据导入、查询分析、数据管理、运维监控、扩展功能等方面。
379 2
云原生数据仓库AnalyticDB产品使用合集之是否可以创建表而不使用分区
|
15天前
|
Cloud Native 关系型数据库 MySQL
《阿里云产品四月刊》—云原生数据仓库 AnalyticDB MySQL 版 新功能
阿里云瑶池数据库云原生化和一体化产品能力升级,多款产品更新迭代
|
1月前
|
SQL Cloud Native 关系型数据库
云原生数据仓库AnalyticDB产品使用合集之如何进行一键诊断
阿里云AnalyticDB提供了全面的数据导入、查询分析、数据管理、运维监控等功能,并通过扩展功能支持与AI平台集成、跨地域复制与联邦查询等高级应用场景,为企业构建实时、高效、可扩展的数据仓库解决方案。以下是对AnalyticDB产品使用合集的概述,包括数据导入、查询分析、数据管理、运维监控、扩展功能等方面。
379 7
|
1月前
|
存储 SQL Cloud Native
云原生数据仓库AnalyticDB产品使用合集之热数据存储空间在什么地方查看
阿里云AnalyticDB提供了全面的数据导入、查询分析、数据管理、运维监控等功能,并通过扩展功能支持与AI平台集成、跨地域复制与联邦查询等高级应用场景,为企业构建实时、高效、可扩展的数据仓库解决方案。以下是对AnalyticDB产品使用合集的概述,包括数据导入、查询分析、数据管理、运维监控、扩展功能等方面。