# 1. 下载Doris
cd /bigdata/app
wget https://apache-doris-releases.oss-accelerate.aliyuncs.com/apache-doris-2.0.3-bin-x64.tar.xz
tar -xvf apache-doris-2.0.3-bin-x64.tar.xz
ln -s apache-doris-2.0.3-bin-x64 doris

# 2. 创建目录
mkdir -p /bigdata/fgdata/doris/meta
mkdir -p /bigdata/fgdata/doris/storage
mkdir -p /bigdata/fgdata/logs/doris

# 3. 配置FE
cat > /bigdata/app/doris/fe/conf/fe.conf << ‘EOF’
meta_dir = /bigdata/fgdata/doris/meta
http_port = 8030
rpc_port = 9020
query_port = 9030
edit_log_port = 9010
EOF

# 4. 启动FE
/bigdata/app/doris/fe/bin/start_fe.sh –daemon

# 5. 验证FE
jps | grep PaloFe
curl http://localhost:8030

# 6. 配置BE
cat > /bigdata/app/doris/be/conf/be.conf << ‘EOF’
storage_root_path = /bigdata/fgdata/doris/storage
be_port = 9060
webserver_port = 8040
heartbeat_service_port = 9050
brpc_port = 8060
EOF

# 7. 添加BE到FE
mysql -h 127.0.0.1 -P 9030 -uroot
mysql> ALTER SYSTEM ADD BACKEND “fgedu-be01:9050”;
mysql> ALTER SYSTEM ADD BACKEND “fgedu-be02:9050”;
mysql> ALTER SYSTEM ADD BACKEND “fgedu-be03:9050”;

# 8. 启动BE
/bigdata/app/doris/be/bin/start_be.sh –daemon

# 9. 验证集群
mysql -h 127.0.0.1 -P 9030 -uroot
mysql> SHOW PROC ‘/backends’;

# 连接Doris
mysql -h fgedu-fe -P 9030 -uroot

# 创建数据库
mysql> CREATE DATABASE IF NOT EXISTS fgedu_db;
mysql> USE fgedu_db;

# 创建Duplicate表（用户行为日志）
mysql> CREATE TABLE IF NOT EXISTS fgedu_user_log (
dt DATE,
ts DATETIME,
user_id BIGINT,
event_type STRING,
page STRING,
duration INT
)
DUPLICATE KEY(dt, ts, user_id)
PARTITION BY RANGE(dt) (
PARTITION p202404 VALUES LESS THAN (‘2024-05-01’)
)
DISTRIBUTED BY HASH(user_id) BUCKETS 32
PROPERTIES (
“replication_num” = “3”
);

# 创建Aggregate表（销售报表）
mysql> CREATE TABLE IF NOT EXISTS fgedu_sales_agg (
dt DATE,
city STRING,
category STRING,
sales_cnt BIGINT SUM,
sales_amt DECIMAL(18,2) SUM
)
AGGREGATE KEY(dt, city, category)
PARTITION BY RANGE(dt) (
PARTITION p202404 VALUES LESS THAN (‘2024-05-01’)
)
DISTRIBUTED BY HASH(city) BUCKETS 16
PROPERTIES (
“replication_num” = “3”
);

# Stream Load导入数据
curl –location-trusted -u root: \
-H “label:fgedu_label_$(date +%Y%m%d_%H%M%S)” \
-H “column_separator:,” \
-T /bigdata/fgdata/user_log.csv \
http://fgedu-fe:8030/api/fgedu_db/fgedu_user_log/_stream_load

# Broker Load从HDFS导入
mysql> LOAD LABEL fgedu_db.label_001 (
DATA INFILE(‘hdfs://fgedu-nn:8020/bigdata/fgdata/hive/warehouse/fgedu_user/*’)
INTO TABLE fgedu_user_log
COLUMNS TERMINATED BY ‘,’
)
WITH BROKER hdfs_broker (
‘username’=’hdfs’,
‘password’=”
)
PROPERTIES (
‘timeout’=’3600’
);

# Spark Doris Connector
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>doris-spark-connector_2.12</artifactId>
<version>1.2.0</version>
</dependency>

# Spark写入Doris
import org.apache.spark.sql.SparkSession

object FgeduSpark2Doris {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder()
.appName(“FgeduSpark2Doris”)
.enableHiveSupport()
.getOrCreate()

val df = spark.read.table(“hive_db.fgedu_user_table”)

df.write
.format(“doris”)
.option(“doris.fenodes”, “fgedu-fe:8030”)
.option(“doris.table.identifier”, “fgedu_db.fgedu_user_table”)
.option(“user”, “root”)
.option(“password”, “”)
.option(“sink.batch.size”, “100000”)
.option(“sink.max-retries”, “3”)
.mode(“append”)
.save()

spark.stop()
}
}

# 提交Spark任务
spark-submit –class FgeduSpark2Doris \
–master yarn \
–deploy-mode cluster \
–jars doris-spark-connector_2.12-1.2.0.jar \
fgedu-spark-doris.jar