Massive data! Second level analysis! Flink+doris build a real-time data warehouse scheme

One 、 background

With the rapid development of business , To meet the real-time report statistics and decision analysis of billions of data , We chose Flink + Doris Real time data warehouse scheme .

Two 、Doris The basic principle

Doris The basic architecture is very simple , Only FE(Frontend)、BE(Backend) Two characters , Does not rely on any external components , Very friendly to deployment and O & M . The architecture is as follows

FE（Frontend） With  Java  Language first .

Main functional responsibilities ：

1）  Receive user connection request （MySql  Protocol layer ）

2）  Metadata storage and management

3）  Query statement parsing and execution plan distribution

4）  Cluster management and control

FE  There are two main roles , One is  follower, One more  observer,leader It is a special election follower.follower It is mainly used to achieve high availability of metadata , Ensure single node downtime , Metadata can be recovered online in real time , Without affecting the whole service .

BE（Backend）  With  C++  Language first .

Main functional responsibilities ：

1）  Data storage and management

2）  Query plan execution

3、 ... and 、 Technical framework

The overall data link is shown in the following figure

1.  adopt FlinkCDC collection Mysql Binlog To Kafka Medium Topic1

2.  Development Flink Tasks consume the above Binlog Generate a wide table of related topics , write in Topic2

3.  To configure Doris Routine Load  Mission , take Topic2 Data import Doris

Four 、 practice

About the steps 1 And steps 2 Practice , In the foreword  “ be based on Flink-CDC Data synchronization ⽅ case ”  It has been explained in the article of , This article will describe the steps 3 Expand the detailed description .

4.1 Build table

Because business data is often accompanied by UPDATE,DELETE  Wait for the operation , In order to keep the data granularity of the real-time data warehouse consistent with the business database , So choose Doris Unique  Model （ The data model is highlighted below ） The specific table creation statements are as follows

CREATE TABLE IF NOT EXISTS table_1
(
key1 varchar(32),
key2 varchar(32),
key3 varchar(32),
value1 int,
value2 varchar(128),
value3 Decimal(20, 6),
data_deal_datetime DateTime COMMENT ' Data processing time ',
data_status INT COMMENT ' Whether the data is deleted ,1 Is normal ,-1 Indicates that the data has been deleted '
) 
ENGINE=OLAP
UNIQUE KEY(`key1`,`key2`,`key3`)
COMMENT "xxx"
DISTRIBUTED BY HASH(`key2`) BUCKETS 32
PROPERTIES (
 "storage_type"="column",
 "replication_num" = "3",
 "function_column.sequence_type" = 'DateTime'
);

You can see , There are two fields in the table structure data_deal_datetime,data_status.

data_deal_datetime Mainly the same key The judgment basis of data coverage in case of

data_status It is compatible with the deletion of data by the business library

4.2 Data import task

Doris Provides active pull Kafka The function of data , The configuration is as follows

CREATE ROUTINE LOAD database.table1 ON table1
COLUMNS(key1,key2,key3,value1,value2,value3,data_deal_datetime,data_status),
ORDER BY data_deal_datetime
PROPERTIES
(
"desired_concurrent_number"="3",
"max_batch_interval" = "10",
"max_batch_rows" = "500000",
"max_batch_size" = "209715200",
"format" = "json",
"json_root" = "$.data",
"jsonpaths"="[\"$.key1\",\"$.key2\",\"$.key3\",\"$.value1\",\"$.value2\",
 \"$.value3\",\"$.data_deal_datetime\",\"$.data_status\"]"
)FROM KAFKA
(
"kafka_broker_list"="broker1_ip:port1,broker2_ip:port2,broker3_ip:port3",
"kafka_topic"="topic_name",
"property.group.id"="group_id",
"property.kafka_default_offsets"="OFFSET_BEGINNING"
);

In the import statement ,

ORDER BY data_deal_datetime  Express basis data_deal_datetime Field to overwrite key Same data

desired_concurrent_number  Indicates the desired concurrency .

max_batch_interval/max_batch_rows/max_batch_size  These three parameters represent

1） Maximum execution time of each subtask .

2） Maximum number of rows read per subtask .

3） Maximum number of bytes read per subtask .

4.3 Task monitoring and alarm

Doris routine load  If dirty data is encountered, the task will be suspended , Therefore, it is necessary to regularly monitor the status of the data import task and automatically recover the failed task . And send the error message to the designated lark Group . The specific script is as follows

import pymysql # Import  pymysql
import requests,json


# Open database connection
db= pymysql.connect(host="host",user="user",
 password="passwd",db="database",port=port)

#  Use cursor() Method get operation cursor
cur = db.cursor()

#1. Query operation
#  To write sql  Query statement  
sql = "show routine load"
cur.execute(sql) # perform sql sentence
results = cur.fetchall() # Get all the records of the query
for row in results :
 name = row[1]
 state = row[7]
 if state != 'RUNNING':
 err_log_urls = row[16]
 reason_state_changed = row[15]
 msg = "doris  The data import task is abnormal :\n name=%s \n state=%s \n reason_state_changed=%s \n err_log_urls=%s \n About to automatically resume , Please check the error message " % (name, state,
reason_state_changed, err_log_urls)
 payload_message = {
 "msg_type": "text",
 "content": {
 "text": msg
 }
}
 url = 'lark  Call the police url'
 s = json.dumps(payload_message)
 r = requests.post(url, data=s)
 cur.execute("resume routine load for " + name)

cur.close()
db.close()

Now online configuration monitoring 1 Once per minute , If a task is suspended , The import task will be resumed automatically , But the dirty data that causes the task to fail will be skipped , At this time, it is necessary to manually check the failure reason , Trigger the import of this data again after repair .

4.4 Data model

Doris Internal table , There are mainly 3 A data model , Namely   Aggregate ,Unique ,Duplicate

Before introducing the data model , Explain first Column, stay Doris in ,Column  It can be divided into two categories ：Key  and  Value. From a business perspective ,Key  and  Value  Corresponding to dimension column and indicator column respectively .

4.4.1Aggregate Model

Simply speaking ,Aggregate A model is a prepolymerization model , Be similar to Molap, By defining in advance key Column and value How columns are aggregated , When data is imported, the key Column the same data according to value Columns are aggregated together , That is, the final exterior and interior key Only one piece of the same data is reserved ,value Calculate according to the corresponding rules . Here's an example .

The table structure is as follows ：

CREATE TABLE tmp_table_1
 (
 user_id varchar(64) COMMENT " user id",
 channel varchar(64) COMMENT " User source channel ",
 city_code varchar(64) COMMENT " User's city code ",
 last_visit_date DATETIME REPLACE DEFAULT "1970-01-01 00:00:00" COMMENT " Last time the user visited ",
 total_cost BIGINT SUM DEFAULT "0" COMMENT " Total user consumption "
 )
ENGINE=OLAP
AGGREGATE KEY(user_id, channel, city_code)
DISTRIBUTED BY HASH(user_id) BUCKETS 6
 PROPERTIES("storage_type"="column","replication_num" = "1");

In the table structure ,key The columns are user_id, channel, city_code ,value The column is last_visit_date,total_cost, Their aggregation methods are REPLACE,SUM.

Now? , Insert a batch of data into the table

insert into tmp_table_1 values('suh_001','JD','001','2022-01-01 00:00:01','57');
insert into tmp_table_1 values('suh_001','JD','001','2022-02-01 00:00:01','76');
insert into tmp_table_1 values('suh_001','JD','001','2022-03-01 00:00:01','107');

According to our understanding , Now? tmp_table_1 Although we inserted 3 Data , But these three data key It's all consistent , Then there should be only one piece of data in the final table , also last_visit_date The value should be "2022-03-01 00:00:01",total_cost The value should be  240. Let's verify

You can see , The result is as we expected ⼀ Cause .

4.4.2Unique  Model

Just like the real-time data warehouse constructed this time , We pay more attention to how to ensure the uniqueness of the primary key ⼀ sex , That is, how to get  Primary Key  only ⼀ Sexual constraints .⼤ Home can refer to ⾯4.1  Examples of creating tables ⼦, Here ⾥ No more examples .

4.4.3Duplicate  Model

In some multidimensional analysis scenarios , The data has no primary key , There are no aggregate requirements . So quote ⼊ Duplicate  The data model is full ⾜ This kind of demand . Illustrate with examples .

The table structure is as follows

CREATE TABLE tmp_table_2
 (
 user_id varchar(64) COMMENT " user id",
 channel varchar(64) COMMENT " User source channel ",
 city_code varchar(64) COMMENT " User's city code ",
 visit_date DATETIME COMMENT " User login time ",
cost BIGINT COMMENT " User consumption amount "
 )
ENGINE=OLAP
DUPLICATE KEY(user_id, channel, city_code)
DISTRIBUTED BY HASH(user_id) BUCKETS 6
 PROPERTIES("storage_type"="column","replication_num" = "1");

insert data

insert into tmp_table_2 values('suh_001','JD','001','2022-01-01 00:00:01','57');
insert into tmp_table_2 values('suh_001','JD','001','2022-02-01 00:00:01','76');
insert into tmp_table_2 values('suh_001','JD','001','2022-03-01 00:00:01','107');


Because the data is Duplicate  Model , Therefore, no processing will be carried out , The query should find 3 Data

4.4.4 Suggestions on the selection of data model

Because the data model has been determined when creating tables , And cannot be modified . therefore , Choosing an appropriate data model is very important .

Aggregate  The model can be pre aggregated , It greatly reduces the amount of data to be scanned and the amount of calculation of queries when aggregating queries , It is very suitable for report query scenarios with fixed patterns . But the model is right  count(*)  The query is very unfriendly . At the same time, because it is fixed  Value  The aggregation method on the column , When performing other types of aggregate queries , Semantic correctness needs to be considered .

Unique  The model is for scenarios that require unique primary key constraints , It can guarantee the uniqueness of the primary key . But you can't take advantage of  ROLLUP  Query advantages brought by pre aggregation .

Duplicate  Suitable for any dimension  Ad-hoc  Inquire about . Although it is also impossible to take advantage of the characteristics of prepolymerization , But not constrained by the aggregation model , It can give play to the advantages of the inventory model .

5、 ... and 、 summary

Flink + Doris  After the built real-time data warehouse goes online , The corresponding speed of the report interface has been significantly improved , Single table 10 Billion level aggregate query response speed TP95 by 0.79 second ,TP99 by 5.03 second .

up to now , The whole digital warehouse system has been running smoothly 8 More than a month .

6、 ... and 、 Reference material

https://doris.apache.org/

About lingchuang group （Advance Intelligence Group）

Lingchuang group was founded in  2016  year , Committed to localized application through scientific and technological innovation , Transform and reshape the financial and retail industries , Create a service for consumers with diversified business layout 、 The ecosystem of enterprises and merchants . The group consists of two parts: enterprise business and consumer business , Enterprise business includes  ADVANCE.AI  and  Ginee, They are banks 、 Finance 、 Financial technology 、 Retail and e-commerce industry customers provide services based on  AI  Digital authentication technology 、 Risk management products and omni channel e-commerce service solutions ; Consumer business  Atome Financial  Including Asia's leading pay after pay platform  Atome  And digital financial services .

2021  year  9  month , Lingchuang group announced the completion of super  4  Billion dollars  D  Round of funding , After the completion of financing, the valuation of lingchuang group has exceeded  20  Billion dollars , Become one of the largest independent technology start-ups in Singapore .

Looking back  BREAK AWAY

Spring data JPA  Practice and principle analysis

How to solve the problem of... In the scenario of massive data update  Mysql  The deadlock problem

Enterprise class  APIs  Safety practice guide  （ It is suggested that junior and intermediate engineers collect ）

Cypress UI  Automated testing framework

serverless Make our operation and maintenance easier

▼  If you think this article is helpful to you , Inspired , Welcome to my collection ：

1、

give the thumbs-up 、 Focus on

Lingchuang group

, Get the latest technology sharing and company dynamics .

2、 Pay attention to our official account  &  Zhihu No 「 Lingchuang group  Advance Group」 Or visit

Official website

, Learn more about enterprise dynamics .