Bitmap是一種經(jīng)典的數(shù)據(jù)結(jié)構(gòu)秃诵，用于高效地對大量的二進制數(shù)據(jù)進行壓縮存儲和快速查詢庞瘸。Doris支持bitmap數(shù)據(jù)類型焰薄，在Flink計算場景中槽卫，可以結(jié)合Flink doris Connector對bitmap數(shù)據(jù)做計算跟压。
社區(qū)里很多小伙伴在是Doris Flink Connector的時候，不知道怎么寫B(tài)itmap類型的數(shù)據(jù)晒夹，本文將介紹如何使用 Flink Doris Connector 如何將 bitmap 數(shù)據(jù)寫入 Doris 中裆馒。
前置準備
Doris2.0.1的環(huán)境
Flink1.16姊氓，同時將 Doris Flink Connector的Jar包放在<FLINK_HOME>/lib 下面丐怯。
創(chuàng)建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數(shù)據(jù)
這里模擬Flink讀取MySQL數(shù)據(jù)寫入Doris，同時將user_id存儲到bitmap中翔横。
模擬數(shù)據(jù)
創(chuàng)建MySQL表
CREATE TABLE `page_view` (
`id` int NOT NULL,
`dt` int,
`page` varchar(256),
`user_id` int,
PRIMARY KEY (`id`)
);
#模擬數(shù)據(jù)
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數(shù)據(jù)
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寫入數(shù)據(jù)
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 來實現(xiàn)的读跷，同樣 Stream load 在 Connector 里也是一樣適用，我們將這個參數(shù)封裝在了 ：sink.properties 參數(shù)里禾唁，
這里我們可以看到上面這個例子里我們在是 With 屬性里加入了我們 Columns 參數(shù)效览，這里我們配置了列的轉(zhuǎn)換操作，將 user_id 通過 to_bitmap 函數(shù)進行轉(zhuǎn)換荡短，并導入到 Doris 表里丐枉。
查詢結(jié)果
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 模擬數(shù)據(jù)寫入剛才的表中。
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");
}
查詢結(jié)果
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)
