前言

虽然这是一个用CDH搭建起来的集群，本篇前半部分大概展示下CDH搭建过程，以及搭建好后它的样子是什么，但是主要侧重记录单独某一组件发挥的作用

CDH搭建步骤

1. 下载ClouderaManager
   点击进入下载界面

2. 环境准备

• 各个主机名之间设置映射
• 关闭并禁用防火墙
• 关闭SELinux
• 设置SSH免密登陆(所有节点)
• 禁用透明大页(所有节点)
• 修改Linux swappiness参数(所有节点)
• 安装JDK(所有节点)要求使用指定版本的oracle-j2sdk1.8
• 上传JDBC依赖包(所有节点)

3. 安装

• 安装MySQL(master节点）
• 安装Apache Httpd服务(master节点）
• 配置Cloudera Manager安装包yum源(master节点)
• 安装 Cloudera Manager(master节点)

原始数据信息表

后期的操作主要围绕下面这六张表的数据展开

集群配置情况

– cdh0051

[root@cdh0051 ~]# jps
3152 HeadlampServer
3153 Main
1651 Main
108244 Jps
3386 EventCatcherService
3150 Main
3151 AlertPublisher

• cdh0052

[root@cdh0052 ~]# jps
3009 ResourceManager
2641 EmbeddedOozieServer
2642 Application
3011 DataNode
2645 QuorumPeerMain
15366 Jps
3035 NodeManager
2639 NameNode

• cdh0053

[root@cdh0053 ~]# jps
2721 SecondaryNameNode
45733 Jps
2744 DataNode
2713 NodeManager
2633 Kafka
2636 QuorumPeerMain

• cdh0054

[root@cdh0054 ~]# jps
69521 Jps
2625 Kafka
2915 NodeManager
2627 HistoryServer
2708 DataNode
2710 QuorumPeerMain
2631 RunJar
2632 RunJar

1、需求分析（梳理表结构，梳理其度量）

度量 Metric

客户数（Customer 数量）
活跃率（活跃客户数/总客户数；活跃定义：一周内下过订单；）
销售额（orderdetails 表内的unit price * quantity – discount）
订单量（order 的数量）
销售量（产品所有order 的quantity 的总和）
销售额Top100（片区）
销售额Topn（员工）
销售额Topn（产品）

维度 Dimension

时间（如按时间切销售额等指标）
员工（如分析不同员工促成的销售额等指标）
销售片区（分析不同销售片区促成的销售额等指标）
产品（分析不同产品的销售额等指标）
产品类别（分析不同类别产品的销售额等指标）

• 以下需求是一个建议，可以从很多角度对数据进行分析

1. 昨日销售额多少？
2. 昨日新增客户多少？
3. 本周总共新增客户多少？
4. 目前平台活跃客户率如何？
5. 平台件均价多少？
6. 本周平台最畅销的产品是什么？意图（度量）：最畅销产品Top-1维度：时间、（销量）
7. 最畅销产品排名前十是？

2、集群架构设计（使用那些组件、所有组件和机器的对应关系）

Cloudera Manager:http://192.168.3.51:7180/
Namenode http://192.168.3.52:9870/
Hue http://192.168.3.53:8889/

cdh0051

cdh0052

cdh0053

cdh0054

3、数据处理流程设计（数据流程图）

处理前先了解各个组件的功能

1. 通过flume把一个kafka集群的数据采集到自己的kafka集群

点击查看flume官方配置文件
http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html#file-channel

a1.sources=r1  
a1.sinks=k1        
a1.channels=c1
 
# configure source 
# source 为kafka类型  
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
//批量大小-一次
a1.sources.r1.batchSize = 5000
//批处理时间-每隔多少毫秒处理的时间
a1.sources.r1.batchDurationMillis = 2000
a1.sources.r1.kafka.bootstrap.servers = cdh0203:9092,cdh0204:9092,cdh0205:9092
/topic
a1.sources.r1.kafka.topics = topic01-categories
#1earliest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
#latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
a1.sources.r1.kafka.consumer.auto.offset.reset = earliest
//组id,名字随意，同组可消费一个topic
a1.sources.r1.kafka.consumer.group.id = ds001
# configure sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
//自己集群的topic，但尽量与数据源topic一致
a1.sinks.k1.kafka.topic = topic01-categories
a1.sinks.k1.kafka.bootstrap.servers = cdh0053:9092,cdh0054:9092
a1.sinks.k1.kafka.flumeBatchSize = 2000
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.k1.kafka.producer.compression.type = snappy
a1.sinks.kafka-sink.kafka.producer.batch.size = 1048576
# configure channels  
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.byteCapacityBufferPercentage = 20
a1.channels.c1.transactionCapacity = 10000
a1.channels.c1.byteCapacity = 800000
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

2. 把本地kafka集群的数据传到hdfs上

3. 设计hive仓库

• 数据仓库与数据库的区别

数据库是面向事物的，它是针对具体业务在数据库联机的日常操作，数据量小
数据仓库是面向处理的，它是针对具体业务在数据库联机的日常操作，数据量大
数据仓库标准上可以分为四层：ODS（临时存储层）、PDW（数据仓库层）、DM（数据集市层）、APP（应用层）。
ODS:一个用于存储当前需要加载的数据，一个用于存储处理完后的历史数据。
PDW:这一层的数据一般是遵循数据库第三范式的
DW:从数据的时间跨度来说，通常是PDW层的一部分，主要的目的是为了满足用户分析的需求
APP:这层数据是完全为了满足具体的分析需求而构建的数据

• 为什么要对数据仓库分层：

用空间换时间，通过大量的预处理来提升应用系统的用户体验（效率），因此数据仓库会存在大量冗余的数据；
2如果不分层的话，如果源业务系统的业务规则发生变化将会影响整个数据清洗过程，工作量巨大
3通过数据分层管理可以简化数据清洗的过程，因为把原来一步的工作分到了多个步骤去完成，相当于把一个复杂的工作拆成了多个简单的工作，把一个大的黑盒变成了一个白盒，每一层的处理逻辑都相对简单和容易理解，这样我们比较容易保证每一个步骤的正确性，当数据发生错误的时候，往往我们只需要局部调整某个步骤即可。

六张表之间的关系

订单表
MSCK REPAIR TABLE命令主要是用来解决通过hdfs dfs -put或者hdfs api写入hive分区表的数据在hive中无法被查询到的问题。

CREATE EXTERNAL TABLE IF NOT EXISTS orders_ods_t
(
orderid INT COMMENT "订单id",    
customerID INT COMMENT "客户id",
employID INT COMMENT "员工id",
orderdate DATE COMMENT "订单生成时间",
requiredDate DATE COMMENT "需用日期",
shippedDate DATE COMMENT "发货日期",
shipvia INT COMMENT "发货方式",
freight DOUBLE COMMENT "运费",
shipname STRING COMMENT "运输方式",
shipadress STRING COMMENT "发货地址",
shipcity STRING COMMENT "发货城市",
shipregion STRING COMMENT "发货区域",
shippostalcode STRING COMMENT "发货邮政编码",
shipcountry STRING COMMENT "发货国家" 
)
COMMENT "订单表"
partitioned by (yt string,mt string,dt string)
row format delimited fields terminated by '\t'
LOCATION '/user/hive/warehouse/ods_database.db/orders_ods_t';
MSCK REPAIR TABLE orders_ods_t;

订单详情表

CREATE EXTERNAL TABLE IF NOT EXISTS orderdetails_ods_t
(
orderid INT COMMENT "订单id",
productId INT COMMENT "产品id",
unitPrice DOUBLE COMMENT "单价",
quantity INT COMMENT "数量",
discount DOUBLE COMMENT "折扣" 
)
COMMENT "订单详情"
partitioned by (yt string,mt string,dt string)
row format delimited fields terminated by '\t'
LOCATION '/user/hive/warehouse/ods_database.db/orderdetails_ods_t';


MSCK REPAIR TABLE orderdetails_ods_t;

顾客表

CREATE EXTERNAL TABLE IF NOT EXISTS customers_ods_t
(
customerID INT COMMENT "客户id",    
companyName STRING COMMENT "公司名称",
contactName STRING COMMENT "联系人姓名",
contactTile STRING COMMENT "联络人职位",
address STRING COMMENT "地址",
city STRING COMMENT "城市",
regdate DATE COMMENT "客户注册时间",
region STRING COMMENT "客户区域",
postalcode STRING COMMENT "邮政编码",
country STRING COMMENT "国家",
phone STRING COMMENT "电话",
fax STRING COMMENT "传真" 
)
COMMENT "顾客表"
partitioned by (yt string,mt string,dt string)
row format delimited fields terminated by '\t'
LOCATION '/user/hive/warehouse/ods_database.db/customers_ods_t';

MSCK REPAIR TABLE customers_ods_t;

员工表

CREATE EXTERNAL TABLE IF NOT EXISTS employees_ods_t
(
employeeId INT COMMENT "员工id",
lastname STRING COMMENT "员工名",
firstName STRING COMMENT "员工姓",
title STRING COMMENT "头衔",
titleofCountry STRING COMMENT "所属国家",
birthdate DATE COMMENT "出生日期",
hiredate DATE COMMENT "入职时间",
address STRING COMMENT "地址",
city STRING COMMENT "城市",
region STRING COMMENT "片区",
postalCode STRING COMMENT "邮编",
country STRING COMMENT "国籍",
homePhone STRING COMMENT "电话",
extension STRING COMMENT "附加",
photo STRING COMMENT "照片",
note STRING COMMENT "注意",
reportTo INT COMMENT "上级领导",
salary DOUBLE COMMENT "工资"
)
COMMENT "员工表"
partitioned by (yt string,mt string,dt string)
row format delimited fields terminated by '\t'
LOCATION '/user/hive/warehouse/ods_database.db/employees_ods_t';

MSCK REPAIR TABLE employees_ods_t;

产品表

CREATE EXTERNAL TABLE IF NOT EXISTS products_ods_t
(
productid INT COMMENT "产品id",
produceName STRING COMMENT "产品名",
supplier INT COMMENT "供应商",
categoryID INT COMMENT "产品类型ID",
quantityPerUint STRING COMMENT "每种产品的数量",
unitPrice DOUBLE COMMENT "单价" 
)
COMMENT "产品表"
partitioned by (yt string,mt string,dt string)
row format delimited fields terminated by '\t'
LOCATION '/user/hive/warehouse/ods_database.db/products_ods_t';

MSCK REPAIR TABLE products_ods_t;
SELECT * FROM products_ods_t LIMIT 5;


partitioned by (dt string,type string)                 //制定分区
row format delimited fields terminated by '\t'         //指定字段分隔符为tab
collection items terminated by ','                     //指定数组中字段分隔符为逗号
map keys terminated by ':'                             //指定字典中KV分隔符为冒号
lines terminated by '\n'                               //指定行分隔符为回车换行
stored as textfile                                     //指定存储类型为文件

产品目录表

CREATE EXTERNAL TABLE IF NOT EXISTS category_ods_t
(
categoryID INT COMMENT "产品类别ID",
categoryName STRING COMMENT "产品类别名称"
)
COMMENT "产品目录表"
partitioned by (year string,month string,day string)
row format delimited fields terminated by '\t'
LOCATION '/user/hive/warehouse/ods_database.db/category_ods_t';

加载hdfs数据到hive

LOAD DATA INPATH 'hdfs://wq1:9000/source-six/order' overwrite into table orders_ods_t partition (yt="2021",mt="03",dt="29");

LOAD DATA INPATH 'hdfs://wq1:9000/source-six/orderdetail' overwrite into table orderdetails_ods_t partition (yt="2021",mt="03",dt="29");

LOAD DATA INPATH 'hdfs://wq1:9000/source-six/custom' overwrite into table customers_ods_t partition (yt="2021",mt="03",dt="29");

LOAD DATA INPATH 'hdfs://wq1:9000/source-six/employee' overwrite into table employees_ods_t partition (yt="2021",mt="03",dt="29");

LOAD DATA INPATH 'hdfs://wq1:9000/source-six/products' overwrite into table products_ods_t partition (yt="2021",mt="03",dt="29");

LOAD DATA INPATH 'hdfs://wq1:9000/source-six/categories' overwrite into table category_ods_t partition(year="2021",month="03",day="29");

4. 写HiveSQL语句处理数据
   最畅销商品排名前100的是？(默认是升序)

create table p100 
as
select 
b.producename ,
sum(a.unitprice*quantity*discount) 
as c 
from shoppingmall.orderdetails_ods_t
as a 
inner join
shoppingmall.products_ods_t 
as b 
on a.productid=b.productid 
group by b.producename 
order by c limit 100;

5. sqoop导数据到MySQL

sqoop安装位置/opt/cloudera/parcels/CDH-6.3.2-\1.cdh6.3.2.p0.1605554/lib/sqoop/

配置文件

sqoop export \
--connect jdbc:mysql://192.168.3.51:3306/hive \
--username root \
--password 123456 \
--table north \
--export-dir /user/hive/warehouse/app_database.db/north \
--fields-terminated-by '\001' \
--m 1 

如果出现这种错误，则是数据库相关的错误，比如数据库字段不对，编码格式等；

ERROR tool.ExportTool: Error during export: 
Export job failed!
        at org.apache.sqoop.mapreduce.ExportJobBase.runExport(ExportJobBase.java:445)
        at org.apache.sqoop.manager.SqlManager.exportTable(SqlManager.java:931)
        at org.apache.sqoop.tool.ExportTool.exportTable(ExportTool.java:80)
        at org.apache.sqoop.tool.ExportTool.run(ExportTool.java:99)
        at org.apache.sqoop.Sqoop.run(Sqoop.java:147)
        at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:76)
        at org.apache.sqoop.Sqoop.runSqoop(Sqoop.java:183)
        at org.apache.sqoop.Sqoop.runTool(Sqoop.java:234)
        at org.apache.sqoop.Sqoop.runTool(Sqoop.java:243)
        at org.apache.sqoop.Sqoop.main(Sqoop.java:252)

6. QuickBI上展示的数据
   每个片区销售额排前10的商品

create table sourth 
as  
select
b.producename ,
sum(a.unitprice*quantity*discount) 
as c  
from 
pdw_database.orderdetails_pdw_t 
as a  
inner join 
pdw_database.products_pdw_t 
as b  
on a.productid=b.productid  
inner join
pdw_database.orders_pdw_t 
as e 
on a.orderid=e.orderid 
where e.shipregion='South' 
group by b.producename 
order by c
limit 10