Apache Doris實(shí)時數(shù)據(jù)分析保姆級使用教程
Doris安裝
集群部署
官網(wǎng)下載地址:
https://doris.apache.org/zh-CN/downloads/downloads.html
選擇二進(jìn)制下載，源碼下載需要自己編譯。解壓doris文件:
tar -zxvf apache-doris-1.0.0-incubating-bin.tar.gz -C /opt/module/
集群規(guī)劃
FE部署
修改配置文件vim conf/fe.conf
meta_dir = /opt/module/doris-meta
集群中分發(fā)存儲路徑和FE配置文件，啟動FE。
# 創(chuàng)建meta文件夾存儲路徑
mkdir /opt/module/doris-meta
# 三臺機(jī)器都要執(zhí)行
sh bin/start_fe.sh --daemon
BE部署
修改配置文件vim conf/be.conf
# storage_root_path配置存儲目錄，可以用;來指定多個目錄，每個目錄后可以跟逗號，指定大小默認(rèn)GB
storage_root_path = /opt/module/doris_storage1,10;/opt/module/doris_storage2
集群中分發(fā)存儲路徑和BE配置文件，啟動BE
# 創(chuàng)建storage_root_path存儲路徑
mkdir /opt/module/doris_storage1
mkdir /opt/module/doris_storage2
# 三臺機(jī)器都要執(zhí)行
sh bin/start_be.sh --daemon
訪問Doris PE節(jié)點(diǎn)
doris可以使用mysql客戶端訪問，如果未安裝，則需要安裝mysql-client。
# 第一次訪問不需要密碼，可以自行設(shè)置密碼
mysql -hdoris1 -P 9030 -uroot
# 修改密碼
set password for 'root' = password('root');
添加BE節(jié)點(diǎn)
通過mysql客戶端登入后，添加be節(jié)點(diǎn)，port為be上的heartbeat_service_port端口，默認(rèn)9050
mysql> ALTER SYSTEM ADD BACKEND "hadoop102:9050";
mysql> ALTER SYSTEM ADD BACKEND "hadoop103:9050";
mysql> ALTER SYSTEM ADD BACKEND "hadoop104:9050";
通過mysql客戶端，檢測be節(jié)點(diǎn)狀態(tài),alive必須為true
mysql> SHOW PROC '/backends';
BROKER部署
可選,非必須部署，啟動BROKER
# 三臺集群都要啟動
sh bin/start_broker.sh --daemon
使用mysql客戶端訪問pe,添加broker節(jié)點(diǎn)
mysql> ALTER SYSTEM ADD BROKER broker_name "hadoop102:8000","hadoop103:8000","hadoop104:8000";查看broker狀態(tài)
mysql> SHOW PROC "/brokers";
擴(kuò)容縮容
Doris可以很方便的擴(kuò)容和縮容FE、BE、Broker實(shí)例。通過頁面訪問進(jìn)行監(jiān)控，訪問8030，賬戶為root,密碼默認(rèn)為空不用填寫，除非上述設(shè)置了密碼使用密碼登錄http://hadoop102:8030
FE 擴(kuò)容和縮容
FE 節(jié)點(diǎn)的擴(kuò)容和縮容過程，不影響當(dāng)前系統(tǒng)運(yùn)行。
使用mysql登錄客戶端后，可以使用sql命令查看FE狀態(tài)，目前就一臺FE。
mysql> SHOW PROC '/frontends';
增加FE節(jié)點(diǎn)，F(xiàn)E分為Leader,Follower和Observer三種角色。默認(rèn)一個集群只能有一個Leader,可以有多個Follower和Observer.其中Leader和Follower組成一個Paxos選擇組，如果Leader宕機(jī)，則剩下的Follower會成為Leader，保證HA。Observer是負(fù)責(zé)同步Leader數(shù)據(jù)的不參與選舉。如果只部署一個FE，則FE默認(rèn)就是Leader
第一個啟動的FE自動成為Leader。在此基礎(chǔ)上，可以添加若干Follower和Observer。添加Follower或Observer。使用mysql-client連接到已啟動的FE，并執(zhí)行：在doris2部署Follower，doris3上部署Observer
# 執(zhí)行其中的一個即可，注解如下
# follower/observer_host IP節(jié)點(diǎn)位置
# edit_log_port fe.conf配置文件中可以查詢到
# ALTER SYSTEM ADD FOLLOWER "follower_host:edit_log_port";
ALTER SYSTEM ADD FOLLOWER "hadoop103:9010";
# ALTER SYSTEM ADD OBSERVER "observer_host:edit_log_port";
ALTER SYSTEM ADD OBSERVER "hadoop104:9010";
需要重啟配置節(jié)點(diǎn)的FE，并添加如下參數(shù)啟動
# --helper參數(shù)指定leader地址和端口號
sh bin/start_fe.sh --helper hadoop102:9010 --daemon
sh bin/start_fe.sh --helper hadoop102:9010 --daemon
全部啟動完畢后，再通過mysql客戶端，查看FE狀況
mysql> SHOW PROC '/frontends';
使用以下命令刪除對應(yīng)的FE節(jié)點(diǎn)ALTER SYSTEM DROP FOLLOWER[OBSERVER] "fe_host:edit_log_port";刪除Follower FE時，確保最終剩余的Follower（包括 Leader）節(jié)點(diǎn)為奇數(shù)
ALTER SYSTEM DROP FOLLOWER "hadoop103:9010";
ALTER SYSTEM DROP OBSERVER "hadoop104:9010";
BE 擴(kuò)容和縮容
增加BE節(jié)點(diǎn)，就像上面安裝一樣在mysql客戶端，使用ALTER SYSTEM ADD BACKEND
刪除BE節(jié)點(diǎn)，使用ALTER SYSTEM DROP BACKEND "be_host:be_heartbeat_service_port";
具體文檔請查看官網(wǎng)。
Doris操作手冊
創(chuàng)建用戶
# 連接doris
mysql -hhadoop102 -P 9030 -uroot
# 創(chuàng)建用戶
mysql> create user 'test' identified by 'test';
# 退出使用test即可登錄
mysql> exit;
mysql -hhadoop102 -P 9030 -utest -ptest
表操作
# 創(chuàng)建數(shù)據(jù)庫
mysql> create database test_db;
# 賦予test用戶test庫權(quán)限
mysql> grant all on test_dn to test;
# 使用數(shù)據(jù)庫
mysql> use test_db;
分區(qū)表
分區(qū)表分為單分區(qū)和復(fù)合分區(qū)
單分區(qū)表，建立一張student表。分桶列為id,桶數(shù)為10，副本數(shù)為1
CREATE TABLE student
(
id INT,
name VARCHAR(50),
age INT,
count BIGINT SUM DEFAULT '0'
)
AGGREGATE KEY (id,name,age)
DISTRIBUTED BY HASH(id) buckets 10
PROPERTIES("replication_num" = "1");
復(fù)合分區(qū)表，第一級稱為Partition,即分區(qū)。用戶指定某一維度列做為分區(qū)列（當(dāng)前只支持整型和時間類型的列），并指定每個分區(qū)的取值范圍。第二級稱為Distribution,即分桶。用戶可以指定一個或多個維度列以及桶數(shù)進(jìn)行HASH分布
#創(chuàng)建student2表，使用dt字段作為分區(qū)列，并且創(chuàng)建3個分區(qū)發(fā)，分別是:
#P202007 范圍值是是小于2020-08-01的數(shù)據(jù)
#P202008 范圍值是2020-08-01到2020-08-31的數(shù)據(jù)
#P202009 范圍值是2020-09-01到2020-09-30的數(shù)據(jù)
CREATE TABLE student2
(
dt DATE,
id INT,
name VARCHAR(50),
age INT,
count BIGINT SUM DEFAULT '0'
)
AGGREGATE KEY (dt,id,name,age)
PARTITION BY RANGE(dt)
(
PARTITION p202007 VALUES LESS THAN ('2020-08-01'),
PARTITION p202008 VALUES LESS THAN ('2020-09-01'),
PARTITION p202009 VALUES LESS THAN ('2020-10-01')
)
DISTRIBUTED BY HASH(id) buckets 10
PROPERTIES("replication_num" = "1");
復(fù)合分區(qū)表，第一級稱為Partition,即分區(qū)。用戶指定某一維度列做為分區(qū)列（當(dāng)前只支持整型和時間類型的列），并指定每個分區(qū)的取值范圍。第二級稱為Distribution,即分桶。用戶可以指定一個或多個維度列以及桶數(shù)進(jìn)行HASH分布.
#創(chuàng)建student2表，使用dt字段作為分區(qū)列，并且創(chuàng)建3個分區(qū)發(fā)，分別是:
#P202007 范圍值是是小于2020-08-01的數(shù)據(jù)
#P202008 范圍值是2020-08-01到2020-08-31的數(shù)據(jù)
#P202009 范圍值是2020-09-01到2020-09-30的數(shù)據(jù)
CREATE TABLE student2
(
dt DATE,
id INT,
name VARCHAR(50),
age INT,
count BIGINT SUM DEFAULT '0'
)
AGGREGATE KEY (dt,id,name,age)
PARTITION BY RANGE(dt)
(
PARTITION p202007 VALUES LESS THAN ('2020-08-01'),
PARTITION p202008 VALUES LESS THAN ('2020-09-01'),
PARTITION p202009 VALUES LESS THAN ('2020-10-01')
)
DISTRIBUTED BY HASH(id) buckets 10
PROPERTIES("replication_num" = "1");
數(shù)據(jù)模型
AGGREGATE KEY
AGGREGATE KEY相同時，新舊記錄將會進(jìn)行聚合操作
AGGREGATE KEY模型可以提前聚合數(shù)據(jù)，適合報表和多維度業(yè)務(wù)
UNIQUE KEY
UNIQUE KEY相同時，新記錄覆蓋舊記錄。目前UNIQUE KEY和AGGREGATE KEY的REPLACE聚合方法一致。適用于有更新需求的業(yè)務(wù)。
DUPLICATE KEY
只指定排序列，相同的行并不會合并。適用于數(shù)據(jù)無需提前聚合的分析業(yè)務(wù)
數(shù)據(jù)導(dǎo)入
為適配不同的數(shù)據(jù)導(dǎo)入需求，Doris系統(tǒng)提供5種不同的導(dǎo)入方式。每種導(dǎo)入方式支持不同的數(shù)據(jù)源，存在不同的方式（異步、同步）
Broker load
Broker load是一個導(dǎo)入的異步方式，支持的數(shù)據(jù)源取決于Broker進(jìn)程支持的數(shù)據(jù)源
基本原理：用戶在提交導(dǎo)入任務(wù)后，F(xiàn)E（Doris系統(tǒng)的元數(shù)據(jù)和調(diào)度節(jié)點(diǎn)）會生成相應(yīng)的PLAN(導(dǎo)入執(zhí)行計劃，BE會執(zhí)行導(dǎo)入計劃將輸入導(dǎo)入Doris中)并根據(jù)BE（Doris系統(tǒng)的計算和存儲節(jié)點(diǎn)）的個數(shù)和文件的大小，將Plan分給多個BE執(zhí)行，每個BE導(dǎo)入一部分?jǐn)?shù)據(jù)。BE在執(zhí)行過程中會從Broker拉取數(shù)據(jù)，在對數(shù)據(jù)轉(zhuǎn)換之后導(dǎo)入系統(tǒng)。所有BE均完成導(dǎo)入，由FE最終決定是否導(dǎo)入是否成功。
測試導(dǎo)入HDFS數(shù)據(jù)到Doris
編寫測試文件，上傳到HDFS.
創(chuàng)建doris表，測試導(dǎo)入
CREATE TABLE student
(
id INT,
name VARCHAR(50),
age INT,
count BIGINT SUM DEFAULT '0'
)
AGGREGATE KEY (id,name,age)
DISTRIBUTED BY HASH(id) buckets 10
PROPERTIES("replication_num" = "1");
編寫diros導(dǎo)入sql，更多參數(shù)請看官網(wǎng)
LOAD LABEL test_db.label1
(
DATA INFILE("hdfs://bigdata:8020/student")
INTO TABLE student
COLUMNS TERMINATED BY ","
(id,name,age,count)
SET
(
id=id,
name=name,
age=age,
count=count
)
)
WITH BROKER broker_name
(
"username"="root"
)
PROPERTIES
(
"timeout" = "3600"
);
查看doris導(dǎo)入狀態(tài)
use test_db;
show load;
查看數(shù)據(jù)導(dǎo)入是否成功
Routine Load
例行導(dǎo)入（Routine Load）功能為用戶提供了一種自動從指定數(shù)據(jù)源進(jìn)行數(shù)據(jù)導(dǎo)入的功能
從Kafka導(dǎo)入數(shù)據(jù)到Doris
創(chuàng)建kafka主題
kafka-topics.sh --zookeeper bigdata:2181 --create --replication-factor 1 --partitions 1 --topic test
啟動kafka生產(chǎn)者生產(chǎn)數(shù)據(jù)
kafka-console-producer.sh --broker-list bigdata:9092 --topic test
# 數(shù)據(jù)格式
{"id":"4","name":"czsqhh","age":"18","count":"50"}
在doris中創(chuàng)建對應(yīng)表
CREATE TABLE kafka_student
(
id INT,
name VARCHAR(50),
age INT,
count BIGINT SUM DEFAULT '0'
)
AGGREGATE KEY (id,name,age)
DISTRIBUTED BY HASH(id) buckets 10
PROPERTIES("replication_num" = "1");
創(chuàng)建導(dǎo)入作業(yè),desired_concurrent_number指定并行度
CREATE ROUTINE LOAD test_db.job1 on kafka_student
PROPERTIES
(
"desired_concurrent_number"="1",
"strict_mode"="false",
"format"="json"
)
FROM KAFKA
(
"kafka_broker_list"= "bigdata:9092",
"kafka_topic" = "test",
"property.group.id" = "test"
);
查看作業(yè)狀態(tài)
SHOW ROUTINE LOAD;
控制作業(yè)
STOP ROUTINE LOAD For jobxxx ：停止作業(yè)
PAUSE ROUTINE LOAD For jobxxx：暫停作業(yè)
RESUME ROUTINE LOAD For jobxxx：重啟作業(yè)
數(shù)據(jù)導(dǎo)出
Drois導(dǎo)出數(shù)據(jù)到HDFS
其他參數(shù)詳見官網(wǎng)
EXPORT TABLE test_db.student
PARTITION (student)
TO "hdfs://bigdata:8020/doris/student/"
WITH BROKER broker_name
(
"username" = "root"
);
Doris代碼操作
Spark
引入依賴
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-yarn_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.27</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>1.2.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>2.10.1</version>
</dependency>
<!-- https://mvnrepository.com/artifact/com.alibaba/druid -->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<version>1.1.10</version>
</dependency>
</dependencies>
讀取doris數(shù)據(jù)
object ReadDoris {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setAppName("testReadDoris").setMaster("local[*]")
val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
val df = sparkSession.read.format("jdbc")
.option("url", "jdbc:mysql://bigdata:9030/test_db")
.option("user", "root")
.option("password", "root")
.option("dbtable", "student")
.load()
df.show()
sparkSession.close();
}
}
Flink
引入依賴
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.14.3</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.12</artifactId>
<version>1.14.3</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.12</artifactId>
<version>1.14.3</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.16</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.12</artifactId>
<version>1.14.3</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-elasticsearch7_2.12</artifactId>
<version>1.14.3</version>
</dependency>
<dependency>
<groupId>org.apache.bahir</groupId>
<artifactId>flink-connector-redis_2.12</artifactId>
<version>1.1-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_2.12</artifactId>
<version>1.14.3</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner_2.12</artifactId>
<version>1.14.3</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-common</artifactId>
<version>1.14.3</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-csv</artifactId>
<version>1.14.3</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc_2.12</artifactId>
<version>1.14.3</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.23</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.12</artifactId>
<version>1.14.3</version>
</dependency>
</dependencies>
讀取數(shù)據(jù)
public static void main(String[] args) {
EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().build();
TableEnvironment tEnv = TableEnvironment.create(settings);
String sourceSql = "CREATE TABLE student (\n" +
"`id` Integer,\n" +
"`name` STRING,\n" +
"`age` Integer\n" +
")WITH (\n" +
"'connector'='jdbc',\n" +
"'url' = 'jdbc:mysql://bigdata:9030/test_db',\n" +
"'username'='root',\n" +
"'password'='root',\n" +
"'table-name'='student'\n" +
")";
tEnv.executeSql(sourceSql);
Table table = tEnv.sqlQuery("select * from student");
table.execute().print();
}
