# gmall-flink-2021

**Repository Path**: tangsixiang/gmall-flink-2021

## Basic Information

- **Project Name**: gmall-flink-2021
- **Description**: Flink实时数仓项目
- **Primary Language**: Java
- **License**: GPL-3.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 4
- **Created**: 2022-08-18
- **Last Updated**: 2022-08-18

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

## 项目顺序

## 1. 项目准备工作
1.1 伪分布式hadoop集群、hbase集群、zookeeper集群、kafka集群、nginx、MySQL数据库

| hostname| IP|
|---|---|
| hadoop200| 192.168.80.200 |
| hadoop201| 192.168.80.201 | 
| hadoop202| 192.168.80.202 |
| hadoop203| 192.168.80.203 |

hadoop版本：3.1.3  
hbase版本： 2.4  
kafka版本： 3.0.0

1.2 在00_dbSQL文件夹中，找到对应的建表语句，在MySQL数据库中建表


## 2. gmall-mock
一共包含3个子项目，是模拟数据源的生成项目

### 2.1 gmall-mock-common 
通用工具类

### 2.2 gmall-mock-db
用来随机生成业务数据  
修改application.properties配置文件夹中的数据库配置。  
运行GmallMockDbApplication生成业务数据。  
![业务数据生成](./00_img/buildBusinessData.jpg)

### 2.3 gmall-mock-log
用来随机生成行为日志数据，存档日志，并发送到指定端口  
如果发送到nginx，修改配置文件  
如果发送到springboot程序,需要修改对应的端口。  
![行为数据](./00_img/buildBehaviorData.jpg)
**注意: 运行该程序前，需要先打包gmall-logger-kafka项目上传到服务器，并运行起来。**

## 2. nginx配置
用来接收mock的数据，实现负载，分流到gmall-logger-kafka项目
![nginx配置](./00_img/nginx_conf.jpg)

## 3. gmall-logger-kafka
根据实际情况配置多个节点，将nginx转发的消息发送到kafka节点
（kafka的生产者）

## 4. gmall-flink-cdc (练习CDC)
监控业务数据库MySQL的binlog日志，将业务数据实时写入到Kafka
（alibaba版本的flinkCDC）
```xml
        <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>flink-connector-mysql-cdc</artifactId>
            <version>1.3.0</version>
        </dependency>
```

## 4. gmall-flink-cdc-ververica (练习CDC)
同上面4，只是版本用的是ververica，两个只需要选择一个使用即可。
```xml
       <dependency>
            <groupId>com.ververica</groupId>
            <artifactId>flink-connector-mysql-cdc</artifactId>
            <version>2.1.1</version>
        </dependency>
```

## 5. gmall-phoenix
验证测试phoenix程序，测试用，可忽略。

## 6. gmall-realtime (实时数仓)

### 6.1 ods层
***com.king.app.ods.FlinkCDC***  
监控业务数据库gmall-210325-flink的变化，发送到kafka的TOPIC：ods_base_db  

### 6.2 dwd层
***com.king.app.dwd.BaseDBApp*** 
 
开发稍微复杂点，使用flinkCDC监控配置表table-process。 
消费Kafka的TOPIC：ods_base_db，通过维度表table-process来做分流的判断：  
如果sinkType是Hbase，则写入Phoenix，作为Dim层  
如果sinkType是Kafka，则写入Kafka，作为dwm层  

自动化写入Phoenix中的表：  
![Phoenix表](./00_img/phoenix_table.jpg)

自动化写入Kafka中的Topic： 
![kafka Topic](./00_img/kafka_topic.jpg)

### 6.3 dwm层
***com.king.app.dwm.OrderWideApp***  
**知识点：双流Join**  
通过消费dw_order_info和dwd_order_detail，然后双流join， 
并且从phoenix中关联对应的维度表信息，完成订单的大宽表。


## 7. 其他
flinkcdc读取MySQL，并写入kafka：  
https://www.cnblogs.com/30go/p/15824833.html  
https://www.cnblogs.com/30go/p/15773357.html

mysql权限问题  
https://www.cnblogs.com/30go/p/15808632.html

fllink大厂视频  
https://www.cnblogs.com/30go/p/15857910.html

scala中JSON字符串转Class对象的问题
https://www.cnblogs.com/30go/p/15890329.html

双流join的测试案例
https://www.cnblogs.com/30go/p/15886547.html

状态的超时设置案例
https://www.cnblogs.com/30go/p/15868563.html