# Flink 实战


<!--more-->


## 1.1  需求分析

 抽取（extract）、转换（transform）、加载（load） 

```json
{
    "dt":"2020-08-12 17:19:40",
    "countryCode":"SA",
    "data":[
        {
            "type":"s2",
            "score":0.1,
            "level":"A+"
        },
        {
            "type":"s5",
            "score":0.5,
            "level":"C"
        }
    ]
}
```

areass AREA_US US
areass AREA_CT TW,HK
areass AREA_AR PK,KW,SA
areass AREA_IN IN        

​    dt， 区域（亚洲区) ,data[]

```json
{
    "area":"AREA_AR",
    "dt":"2020-08-12 17:20:25",
    "score":0.8,
    "level":"C",
    "type":"s5"
}
```

## 1.2  项目架构

实现思路：

1、产生数据源

2、将区域和国家的对应关系数据进行保存----Redis

3、通过代码，将redis中的映射关系数据取出，并保存到HashMap中

4、将数据源中的每一条数据发送到Kafka上保存，allData这个Topic

5、通过Flink的消费者Consumer，从allData这个Topic上去拉取数据

6、通过Flink的自定义数据源，从Redis中拉取映射关系的数据

7、在Flink中转化数据源

8、将转化好的数据进行保存 -- kafka allDataDone这个topic

分析：

- flume采集数据
- 将flume采集到的数据发送到Kafaka的名为：AllData的Topic上
- 通过flink读取kafka上alldata的数据，进行实时转换
- 数据解析完成后，通过FlinkKafkaProducer将数据写到Kafka的alldataclean中
- 为了后期的离线操作，通过Flume对数据进行分类落盘操作

代码： [anli](https://gitee.com/ixinglan/flink-demo.git)