怎么使用 Flink 向 Apache Doris 表中写 Bitmap 类型的数据

本文涉及的产品
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
实时计算 Flink 版,5000CU*H 3个月
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
简介: 怎么使用 Flink 向 Apache Doris 表中写 Bitmap 类型的数据

Bitmap是一种经典的数据结构,用于高效地对大量的二进制数据进行压缩存储和快速查询。Doris支持bitmap数据类型,在Flink计算场景中,可以结合Flink doris Connector对bitmap数据做计算。


社区里很多小伙伴在是Doris Flink Connector的时候,不知道怎么写Bitmap类型的数据,本文将介绍如何使用 Flink Doris Connector 如何将 bitmap 数据写入 Doris 中。


前置准备


Doris2.0.1的环境


Flink1.16,同时将 Doris Flink Connector的Jar包放在<FLINK_HOME>/lib 下面。


创建Doris表

CREATE TABLE `page_view_bitmap` (
`dt` int,
`page` varchar(256),
`user_id` bitmap bitmap_union
)
AGGREGATE KEY(`dt`, page)
DISTRIBUTED BY HASH(`dt`) BUCKETS 1
PROPERTIES (
"replication_num" = "1"
)

写入Bitmap数据


这里模拟Flink读取MySQL数据写入Doris,同时将user_id存储到bitmap中。


模拟数据


创建MySQL表

CREATE TABLE `page_view` (
 `id` int NOT NULL,
 `dt` int,
 `page` varchar(256),
 `user_id` int,
 PRIMARY KEY (`id`)
);
#模拟数据
INSERT INTO `test`.`page_view` (`id`, `dt`, `page`, `user_id`) VALUES (1, 20230921, 'home', 1001);
INSERT INTO `test`.`page_view` (`id`, `dt`, `page`, `user_id`) VALUES (2, 20230921, 'home', 1002);
INSERT INTO `test`.`page_view` (`id`, `dt`, `page`, `user_id`) VALUES (3, 20230921, 'search', 1003);
INSERT INTO `test`.`page_view` (`id`, `dt`, `page`, `user_id`) VALUES (4, 20230922, 'mine', 1001);
INSERT INTO `test`.`page_view` (`id`, `dt`, `page`, `user_id`) VALUES (5, 20230922, 'home', 1002);
FlinkSQL写入Bitmap
#使用JDBC读取mysql数据
CREATE TABLE page_view (
   `dt` int,
   `page` string,
   `user_id` int
) WITH (
   'connector' = 'jdbc',
   'url' = 'jdbc:mysql://127.0.0.1:3306/test',
   'table-name' = 'page_view',
   'username' = 'root',
   'password' = '123456'
);

doris connector写入数据

CREATE TABLE page_view_bitmap (
dt int,
page string,
user_id int
)
WITH (
 'connector' = 'doris',
 'fenodes' = '127.0.0.1:8030',
 'table.identifier' = 'test.page_view_bitmap',
 'username' = 'root',
 'password' = '',
 'sink.label-prefix' = 'doris_label1',
 'sink.properties.columns' = 'dt,page,user_id,user_id=to_bitmap(user_id)'
);

insert into page_view_bitmap select * from page_view


我们知道 Doris Flink Connector Sink 底层是基于 Doris Stream Load 来实现的,同样 Stream load 在 Connector 里也是一样适用,我们将这个参数封装在了 :sink.properties 参数里,


这里我们可以看到上面这个例子里我们在是 With 属性里加入了我们 Columns 参数,这里我们配置了列的转换操作,将 user_id 通过 to_bitmap 函数进行转换,并导入到 Doris 表里。


查询结果

mysql> select dt,page,bitmap_to_string(user_id) from `test`.`page_view_bitmap`;
+----------+--------+---------------------------+
| dt       | page   | bitmap_to_string(user_id) |
+----------+--------+---------------------------+
| 20230921 | home   | 1001,1002                 |
| 20230921 | search | 1003                      |
| 20230922 | home   | 1002                      |
| 20230922 | mine   | 1001                      |
+----------+--------+---------------------------+
4 rows in set (0.00 sec)

Flink DataStream


使用 DataStream API 模拟数据写入刚才的表中。


DataStream API 对 Bitmap 的操作也是和上面 SQL 操作的方式一样。

public static void main(String[] args) throws Exception {
       StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
       env.setParallelism(1);
       env.setRuntimeMode(RuntimeExecutionMode.BATCH);
       DorisSink.Builder<String> builder = DorisSink.builder();
       final DorisReadOptions.Builder readOptionBuilder = DorisReadOptions.builder();
       Properties properties = new Properties();
       properties.setProperty("column_separator", ",");
       properties.setProperty("format", "csv");
       properties.setProperty("columns", "dt,page,user_id,user_id=to_bitmap(user_id)");
       DorisOptions.Builder dorisBuilder = DorisOptions.builder();
       dorisBuilder.setFenodes("127.0.0.1:8030")
              .setTableIdentifier("test.page_view_bitmap")
              .setUsername("root")
              .setPassword("");
       DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
       executionBuilder.setLabelPrefix("doris_label")
              .setStreamLoadProp(properties)
              .setDeletable(false);
       builder.setDorisReadOptions(readOptionBuilder.build())
              .setDorisExecutionOptions(executionBuilder.build())
              .setSerializer(new SimpleStringSerializer())
              .setDorisOptions(dorisBuilder.build());
       //mock data
       DataStreamSource<String> stringDataStreamSource = env.fromCollection(
               Arrays.asList("20230921,home,1003", "20230921,search,1001", "20230923,home,1001"));
       stringDataStreamSource.sinkTo(builder.build());
       env.execute("doris bitmap write");
  }

查询结果

mysql> select dt,page,bitmap_to_string(user_id) from `test`.`page_view_bitmap`;
+----------+--------+---------------------------+
| dt       | page   | bitmap_to_string(user_id) |
+----------+--------+---------------------------+
| 20230921 | home   | 1001,1002,1003            |
| 20230921 | search | 1001,1003                 |
| 20230922 | home   | 1002                      |
| 20230922 | mine   | 1001                      |
| 20230923 | home   | 1001                      |
+----------+--------+---------------------------+
5 rows in set (0.00 sec)


相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
相关文章
|
6天前
|
SQL 消息中间件 Java
兼容Trino Connector,扩展Apache Doris数据源接入能力|Lakehouse 使用手册(四)
通过兼容 Connector 插件,Apache Doris 能够支持 Trino/Presto 可对接的所有数据源,而无需改动 Doris 的内核代码。
兼容Trino Connector,扩展Apache Doris数据源接入能力|Lakehouse 使用手册(四)
|
13天前
|
存储 消息中间件 运维
招联金融基于 Apache Doris 数仓升级:单集群 QPS 超 10w,存储成本降低 70%
招联内部已有 40+ 个项目使用 Apache Doris ,拥有超百台集群节点,个别集群峰值 QPS 可达 10w+ 。通过应用 Doris ,招联金融在多场景中均有显著的收益,比如标签关联计算效率相较之前有 6 倍的提升,同等规模数据存储成本节省超 2/3,真正实现了降本提效。
招联金融基于 Apache Doris 数仓升级:单集群 QPS 超 10w,存储成本降低 70%
|
7天前
|
存储 大数据 数据挖掘
【数据新纪元】Apache Doris:重塑实时分析性能,解锁大数据处理新速度,引爆数据价值潜能!
【9月更文挑战第5天】Apache Doris以其卓越的性能、灵活的架构和高效的数据处理能力,正在重塑实时分析的性能极限,解锁大数据处理的新速度,引爆数据价值的无限潜能。在未来的发展中,我们有理由相信Apache Doris将继续引领数据处理的潮流,为企业提供更快速、更准确、更智能的数据洞察和决策支持。让我们携手并进,共同探索数据新纪元的无限可能!
47 11
|
5天前
|
SQL 安全 数据处理
揭秘数据脱敏神器:Flink SQL的神秘力量,守护你的数据宝藏!
【9月更文挑战第7天】在大数据时代,数据管理和处理尤为重要,尤其在保障数据安全与隐私方面。本文探讨如何利用Flink SQL实现数据脱敏,为实时数据处理提供有效的隐私保护方案。数据脱敏涉及在处理、存储或传输前对敏感数据进行加密、遮蔽或替换,以遵守数据保护法规(如GDPR)。Flink SQL通过内置函数和表达式支持这一过程。
24 2
|
13天前
|
关系型数据库 MySQL API
Apache Doris集群部署
Apache Doris集群部署
|
11天前
|
Java 微服务 Spring
驾驭复杂性:Spring Cloud在微服务构建中的决胜法则
【8月更文挑战第31天】Spring Cloud是在Spring Framework基础上打造的微服务解决方案,提供服务发现、配置管理、消息路由等功能,适用于构建复杂的微服务架构。本文介绍如何利用Spring Cloud搭建微服务,包括Eureka服务发现、Config Server配置管理和Zuul API网关等组件的配置与使用。通过Spring Cloud,可实现快速开发、自动化配置,并提升系统的伸缩性和容错性,尽管仍需面对分布式事务等挑战,但其强大的社区支持有助于解决问题。
23 0
|
14天前
|
消息中间件 Kafka 数据处理
实时计算 Flink版产品使用问题之如何处理数据并记录每条数据的变更
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
17天前
|
存储 消息中间件 Java
Apache Flink 实践问题之原生TM UI日志问题如何解决
Apache Flink 实践问题之原生TM UI日志问题如何解决
29 1
|
15天前
|
消息中间件 监控 数据挖掘
基于RabbitMQ与Apache Flink构建实时分析系统
【8月更文第28天】本文将介绍如何利用RabbitMQ作为数据源,结合Apache Flink进行实时数据分析。我们将构建一个简单的实时分析系统,该系统能够接收来自不同来源的数据,对数据进行实时处理,并将结果输出到另一个队列或存储系统中。
59 2

推荐镜像

更多