系列文章目录
广告数仓:采集通道创建
广告数仓:数仓搭建
广告数仓:数仓搭建(二)
广告数仓:全流程调度
文章目录
- 系列文章目录
- 前言
- 一、ClickHouse安装
- 1.修改环境
- 2.安装依赖
- 3.单机安装
- 4.修改配置文件
- 5.启动clickhouse
- 6.创建需要的数据库和表
- 7.Hive数据导出至Clickhouse
- 1.新建IDEA工程
- 2.pom依赖
- 3.HiveToClickhouse.java
- 4.上传hive-site.xml
- 5.打包编译
- 二、DolphinScheduler安装
- 1.软件下载
- 2.编写部署脚本
- 3.创建元数据库及用户
- 4.运行脚本
- 三、全流程调度
- 1.dolphinscheduler启动
- 2.环境准备
- 1.创建租户
- 2.创建用户
- 3.创建环境
- 4.worker分组
- 5.上传脚本
- 6.创建项目
- 7.定义工作流
- 3.数据准备
- 1.配置数据日期
- 2.启动环境
- 3.生成数据
- 4.全流程调度
- 总结
前言
这次我们为广告仓库进行一些收尾工作,主要是最终的数据导出和全流程调度以及数据的可视化,不知道一次能不能写完
一、ClickHouse安装
ClickHouse是一个用于联机分析(OLAP)的列式数据库管理系统(DBMS)。
1.修改环境
sudo vim /etc/security/limits.conf
#####
* soft nofile 65536
* hard nofile 65536
* soft nproc 131072
* hard nproc 131072
sudo vim /etc/security/limits.d/20-nproc.conf
#####
* soft nofile 65536
* hard nofile 65536
* soft nproc 131072
* hard nproc 131072
sudo vim /etc/selinux/config
#####
SELINUX=disabled
分发配置文件
sudo xsync /etc/security/limits.conf
sudo xsync /etc/security/limits.d/20-nproc.conf
sudo xsync /etc/selinux/config
2.安装依赖
三台电脑建议都安装
sudo yum install -y libtool
sudo yum install -y *unixODBC*
3.单机安装
企业实战中,根据自己的需求选择安装的节点数。
这里为了考虑到电脑性能,就是用单机安装。
在software目录下创建clickhouse目录,然后上传提供的安装包
在当前目录下执行所有rpm文件
sudo rpm -ivh *.rpm
4.修改配置文件
sudo vim /etc/clickhouse-server/config.xml
把这一行注释去掉
5.启动clickhouse
sudo systemctl start clickhouse-server
链接测试
clickhouse-client -m
6.创建需要的数据库和表
create database ad_report;
use ad_report;
create table if not exists dwd_ad_event_inc
(
event_time Int64 comment '事件时间',
event_type String comment '事件类型',
ad_id String comment '广告id',
ad_name String comment '广告名称',
ad_product_id String comment '广告产品id',
ad_product_name String comment '广告产品名称',
ad_product_price Decimal(16, 2) comment '广告产品价格',
ad_material_id String comment '广告素材id',
ad_material_url String comment '广告素材url',
ad_group_id String comment '广告组id',
platform_id String comment '推广平台id',
platform_name_en String comment '推广平台名称(英文)',
platform_name_zh String comment '推广平台名称(中文)',
client_country String comment '客户端所处国家',
client_area String comment '客户端所处地区',
client_province String comment '客户端所处省份',
client_city String comment '客户端所处城市',
client_ip String comment '客户端ip地址',
client_device_id String comment '客户端设备id',
client_os_type String comment '客户端操作系统类型',
client_os_version String comment '客户端操作系统版本',
client_browser_type String comment '客户端浏览器类型',
client_browser_version String comment '客户端浏览器版本',
client_user_agent String comment '客户端UA',
is_invalid_traffic UInt8 comment '是否是异常流量'
) ENGINE = MergeTree()
ORDER BY (event_time, ad_name, event_type, client_province, client_city, client_os_type,
client_browser_type, is_invalid_traffic);
7.Hive数据导出至Clickhouse
1.新建IDEA工程
2.pom依赖
<dependencies>
<!-- 引入mysql驱动,目的是访问hive的metastore元数据-->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.33</version>
</dependency>
<!-- 引入spark-hive模块-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.12</artifactId>
<version>3.3.1</version>
<scope>provided</scope>
</dependency>
<!--引入clickhouse-jdbc驱动,为解决依赖冲突,需排除jackson的两个依赖-->
<dependency>
<groupId>ru.yandex.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
<version>0.2.4</version>
<exclusions>
<exclusion>
<artifactId>jackson-databind</artifactId>
<groupId>com.fasterxml.jackson.core</groupId>
</exclusion>
<exclusion>
<artifactId>jackson-core</artifactId>
<groupId>com.fasterxml.jackson.core</groupId>
</exclusion>
</exclusions>
</dependency>
<!-- 引入commons-cli,目的是方便处理程序的输入参数 -->
<dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
<version>1.2</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.0.0</version>
<configuration>
<!--将依赖编译到jar包中-->
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<!--配置执行器-->
<execution>
<id>make-assembly</id>
<!--绑定到package执行周期上-->
<phase>package</phase>
<goals>
<!--只运行一次-->
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
3.HiveToClickhouse.java
package com.atguigu.ad.spark;
import org.apache.commons.cli.*;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
public class HiveToClickhouse {
public static void main(String[] args) {
// 使用common-cli处理传入参数
// 1 定义能够传入那些参数
Options options = new Options();
options.addOption(OptionBuilder.withLongOpt("hive_db").withDescription("hive数据库名称(required)").hasArg(true).isRequired(true).create());
options.addOption(OptionBuilder.withLongOpt("hive_table").withDescription("hive表名称(required)").hasArg(true).isRequired(true).create());
options.addOption(OptionBuilder.withLongOpt("hive_partition").withDescription("hive分区(required)").hasArg(true).isRequired(true).create());
options.addOption(OptionBuilder.withLongOpt("ck_url").withDescription("clickhouse的jdbc url(required)").hasArg(true).isRequired(true).create());
options.addOption(OptionBuilder.withLongOpt("ck_table").withDescription("clickhouse表名称(required)").hasArg(true).isRequired(true).create());
options.addOption(OptionBuilder.withLongOpt("batch_size").withDescription("数据写入clickhouse时的批次大小(required)").hasArg(true).isRequired(true).create());
// 解析参数
GnuParser gnuParser = new GnuParser();
CommandLine cmd = null;
try {
cmd = gnuParser.parse(options, args);
} catch (ParseException e) {
e.printStackTrace();
return;
}
// 创建sparksql环境
SparkConf conf = new SparkConf().setAppName("HiveToClickhouse");
SparkSession sparkSession = SparkSession.builder()
.enableHiveSupport()
.config(conf)
.getOrCreate();
// 读取hive中的数据
sparkSession.sql("set spark.sql.parser.quotedRegexColumnNames=true");
Dataset<Row> dataset = sparkSession
.sql(""+"select `(dt)?+.+` from " + cmd.getOptionValue("hive_db") + "." + cmd.getOptionValue("hive_table") + " where dt= '" + cmd.getOptionValue("hive_partition")+"'");
dataset.show();
// 写入到clickHouse中
dataset.write().mode("append")
.format("jdbc")
.option("url", cmd.getOptionValue("ck_url"))
.option("driver", "ru.yandex.clickhouse.ClickHouseDriver")
.option("dbtable", cmd.getOptionValue("ck_table"))
.option("batch_size", cmd.getOptionValue("batch_size"))
.save();
sparkSession.close();
}
}
4.上传hive-site.xml
5.打包编译
二、DolphinScheduler安装
1.软件下载
阿里源下载
版本号2.0.8
然后解压即可
2.编写部署脚本
vim conf/config/install_config.conf
这里说一下要修改的地方
ips="hadoop102,hadoop103,hadoop104"
# 将要部署任一 DolphinScheduler 服务的服务器主机名或 ip 列表
masters="hadoop102"
# master 所在主机名列表,必须是 ips 的子集
workers="hadoop102:default,hadoop103:default,hadoop104:default"
# worker主机名及队列,此处的 ip 必须在 ips 列表中
alertServer="hadoop102"
# 告警服务所在服务器主机名
apiServers="hadoop102"
# api服务所在服务器主机名
# pythonGatewayServers="ds1"
# 不需要的配置项,可以保留默认值,也可以用 # 注释
installPath="/opt/module/dolphinscheduler"
# DS 安装路径,如果不存在会创建
deployUser="atguigu"
# 部署用户,任务执行服务是以 sudo -u {linux-user} 切换不同 Linux 用户的方式来实现多租户运行作业,因此该用户必须有免密的 sudo 权限。
javaHome="/opt/module/jdk8u282-b08"
# JAVA_HOME 路径
# 数据库信息
DATABASE_TYPE=mysql
SPRING_DATASOURCE_URL=${SPRING_DATASOURCE_URL:-"jdbc:mysql://hadoop102:3306/dolphinscheduler?useUnicode=true&allowPublicKeyRetrieval=true&characterEncoding=UTF-8"}
# Spring datasource username
SPRING_DATASOURCE_USERNAME=${SPRING_DATASOURCE_USERNAME:-"dolphinscheduler"}
# Spring datasource password
SPRING_DATASOURCE_PASSWORD=${SPRING_DATASOURCE_PASSWORD:-"dolphinscheduler"}
registryServers="hadoop102:2181,hadoop103:2181,hadoop104:2181"
# 注册中心地址,即 Zookeeper 集群的地址
registryNamespace="dolphinscheduler"
# DS 在 Zookeeper 的结点名称
resourceStorageType="HDFS"
# 资源存储类型
resourceUploadPath="/dolphinscheduler"
# 资源上传路径
defaultFS="hdfs://hadoop102:8020"
# 默认文件系统
yarnHaIps=
# Yarn RM 高可用 ip,若未启用 RM 高可用,则将该值置空
singleYarnIp="hadoop103"
# Yarn RM 主机名,若启用了 HA 或未启用 RM,保留默认值
hdfsRootUser="atguigu"
# 拥有 HDFS 根目录操作权限的用户
3.创建元数据库及用户
先登录Mysql
CREATE DATABASE dolphinscheduler DEFAULT CHARACTER SET utf8 DEFAULT COLLATE utf8_general_ci;
CREATE USER 'dolphinscheduler'@'%' IDENTIFIED BY 'dolphinscheduler';
GRANT ALL PRIVILEGES ON dolphinscheduler.* TO 'dolphinscheduler'@'%';
flush privileges;
将连接Mysql的Jar拷入lib目录
4.运行脚本
由于dolphinscheduler依赖Hadoop和ZK所以要现将这两个应用都启动了。
myhadoop.sh start
zk.sh start
script/create-dolphinscheduler.sh
./install.sh
初始化安装好后,会直接帮你启动。
用网页端查看以下
hadoop102:12345/dolphinscheduler
初始账号密码
admin
dolphinscheduler123
常用命令
./bin/start-all.sh
./bin/stop-all.sh
./bin/status-all.sh
三、全流程调度
1.dolphinscheduler启动
考虑到虚拟机的资源问题,我们选择测试使用的单机模式,会省一点资源
./bin/dolphinscheduler-daemon.sh start standalone-server
由于我们用的是同一个数据库,所以效果适合集群一样的。
2.环境准备
1.创建租户
注意要和Linux用户相同
2.创建用户
这里要选择我们刚刚创建的租户
3.创建环境
具体参数根据自己电脑修改
export HADOOP_HOME=/opt/module/hadoop
export HADOOP_CONF_DIR=/opt/module/hadoop/etc/hadoop
export SPARK_HOME=/opt/module/spark
export SPARK_HOME1=/opt/module/spark
export SPARK_HOME2=/opt/module/spark
export JAVA_HOME=/opt/module/jdk8u282-b
export HIVE_HOME=/opt/module/hive
export DATAX_HOME=/opt/module/datax
export PATH=$PATH:$HADOOP_HOME/bin:$SPARK_HOME/bin:$JAVA_HOME/bin:$HIVE_HOME/bin:$DATAX_HOME/bin
4.worker分组
因为咱们就一台几点跑DS,所以就写一个worker地址就行。
5.上传脚本
用普通用户上传,否则普通用户无法调用管理员用户上传的文件
创建两个文件夹,分别上传需要的脚本。
6.创建项目
7.定义工作流
名称随意
ad_mysql_to_hdfs_full
hdfs_to_ods
ods_to_dim
ods_to_dwd
HiveToClickhouse
com.atguigu.ad.spark.HiveToClickhouse
--hive_db ad \
--hive_table dwd_ad_event_inc \
--hive_partition ${dt} \
--ck_url jdbc:clickhouse://hadoop102:8123/ad_report \
--ck_table dwd_ad_event_inc \
--batch_size 1000
最后全局保存的时候定义全局变量
3.数据准备
准备数据可以先把DS关掉
启动需要的环境
1.配置数据日期
hadoop102和103修改
vim /opt/module/ad_mock/nginxLogGen.setting
2.启动环境
hadoop.sh start
zk.sh start
kf.sh start
ad_f1.sh start
ad_f2.sh start
3.生成数据
ad_mock.sh
等待一段时间后,查看HDFS
等待数据生成完成即可。
4.全流程调度
现在我们进行全流程调度,可以把两个Flume进程和Kafka进程关掉,然后启动hiveserver2来启动hive
由于我们是本地单节点运行,所以最后把日志一起监控一下,除了问题方便排错。
tail -f logs/dolphinscheduler-standalone-server-hadoop102.out
上线工作流
运行工作流
出错了可以检查日志
我么可以用dbeaver连接clickhouse查看一下
总结
全流程调度就到这里吧,可视化还要再来一次。