Flink数据清洗(Kafka事实表+Redis维度表)

拼搏现实的明天。 2023-01-03 04:43 327阅读 0赞

## 调研 ##

从网上的调研来看，其实整个百度有清洗流程的只有\[1\]其他都是抄的\[1\]中的内容。

### 实验流程 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FwcGxleXVjaGk_size_16_color_FFFFFF_t_70][]

这个流程的话,不要去研究redis的Flink SQL Client的操作方法,因为在mvn repository中

没有看到flink-sql-connector-redis之类 的jar

所以该流程适可而止吧

\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#

### **Redis數據準備** ###

127.0.0.1:6379> hset areas AREA\_US US  
127.0.0.1:6379> hset areas AREA\_CT TW,HK  
127.0.0.1:6379> hset areas AREA\_IN IN  
127.0.0.1:6379> hset areas AREA\_AR PK,SA,KW  
127.0.0.1:6379> hset areas AREA\_IN IN

127.0.0.1:6379> hgetall areas  
1) "AREA\_US"  
2) "US"  
3) "AREA\_CT"  
4) "TW,HK"  
5) "AREA\_IN"  
6) "IN"  
7) "AREA\_AR"  
8) "PK,SA,KW"

本实验的redis对象是没有密码的,如果事先设置了密码,可以根据\[14\]去除

### Redis代码中的注意事项 ###

代码中有这么一句话:

this.jedis = new Jedis("127.0.0.1", 6379);

注意,这里的127.0.0.1如果改成redis所在节点的域名的话,必须是该redis支持外网访问,否则此处不要修改,会导致数据读取失败

\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#

### 本实验注意事项 ###

①redis相关的jar依赖其实目前官方没有在维护了.所以不要做太深入的钻研

②需要导入flink-shaded-hadoop-3-uber-3.1.1.7.0.3.0-79-7.0.jar

Project Structure->Global Libraries中间一列导入上述的jar

否则会报错找不到hdfs这个file system

\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#

### 数据清洗目标 ###

kafka(存放事实表)中数据示范:

\{"dt":"2021-01-11 12:30:32","countryCode":"PK","data":\[\{"type":"s3","score":0.8,"level":"C"\},\{"type":"s5","score":0.1,"level":"C"\}\]\}

格式化后如下:

{
      "dt": "2021-01-11 12:30:32",
      "countryCode": "PK",
      "data": [
        {
          "type": "s3",
          "score": 0.8,
          "level": "C"
        },
        {
          "type": "s5",
          "score": 0.1,
          "level": "C"
        }
      ]
    }

这样的一条数据,根据countryCode转化为redis(存放维度表)中的具体地区AREA\_AR

后面list中的数据打散,最终想要的效果如下:

\{"area":"AREA\_AR","dt":"2021-01-11 12:30:32","score":0.8,"level":"C","type":"s3"\}

\{"area":"AREA\_AR","dt":"2021-01-11 12:30:32","score":0.1,"level":"C","type":"s5"\}

也就是想要根据上述要求,把一条数据转化为两条数据

\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#

### 完整实验操作与代码 ###

[https://gitee.com/appleyuchi/Flink\_Code/tree/master/flink清洗数据案例/FlinkProj][https_gitee.com_appleyuchi_Flink_Code_tree_master_flink_FlinkProj]

\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#

### 可能涉及到的Kafka操作 ###

<table> 
 <tbody> 
 <tr> 
 <td style="width:268px;">操作</td> 
 <td style="width:534px;">命令</td> 
 <td style="width:445px;">备注</td> 
 </tr> 
 <tr> 
 <td style="width:268px;">查看topic</td> 
 <td style="width:534px;">$KAFKA/bin/kafka-topics.sh --list --zookeeper Desktop:2181</td> 
 <td style="width:445px;"> 无 &nbsp; &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:268px;">往allData这个 topic发送 json 消息</td> 
 <td style="width:534px;">$KAFKA/bin/kafka-console-producer.sh --broker-list Desktop:9091 --topic allData</td> 
 <td style="width:445px;"> 这里可能碰到[2]中的报错,注意检查命令中端口与配置文件server.properties中的listeners的端口严格保持一致 [2]中的报错还可能是某个节点的kafka挂掉导致的. &nbsp; 可能碰到[3] 注意关闭防火墙 &nbsp; &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:268px;">使用kafka自带消费端测试下消费</td> 
 <td style="width:534px;">$KAFKA/bin/kafka-console-consumer.sh --bootstrap-server Desktop:9091 --from-beginning --topic allData</td> 
 <td style="width:445px;"> 如果kafka自带消费者测试有问题，那么就不用继续往下面做了， 此时如果使用Flink SQL Client来消费也必然会出现问题 </td> 
 </tr> 
 <tr> 
 <td style="width:268px;">清除topic中所有数据[13]</td> 
 <td style="width:534px;">$KAFKA/bin/kafka-topics.sh --zookeeper Desktop:2181 --delete --topic allData</td> 
 <td style="width:445px;"> 需要$KAFKA/config/server.properties设置 delete.topic.enable=true </td> 
 </tr> 
 </tbody> 
</table>

**Reference:**

**\[1\][【19】Flink 实战案例开发（一）：数据清洗][19_Flink](完整代码+数据，依赖有问题)**

\[2\][Flink清洗Kafka数据存入MySQL测试][Flink_Kafka_MySQL](数据好像不太完整)

**\[3\][Flink案例开发之数据清洗、数据报表展现][Flink](与\[1\]内容重复)**

\[4\][Flink继续实践：从日志清洗到实时统计内容PV等多个指标][Flink_PV](代码不完整)

\[5\][Flink清洗日志服务SLS的数据并求ACU&PCU][Flink_SLS_ACU_PCU](工程文件不完整)

下面的最后考虑(博主说是完整的.下面的实验的原始出处其实是\[1\])

\[6\][Flink入门及实战（20）- 数据清洗实时ETL（1）][Flink_20_- _ETL_1]

\[7\][Flink入门及实战（21）- 数据清洗实时ETL（2）][Flink_21_- _ETL_2]

\[8\][Flink入门及实战（22）- 数据清洗实时ETL（3）][Flink_22_- _ETL_3]\[10\][Flink 清理过期 Checkpoint 目录的正确姿势][Flink _ Checkpoint]

\[11\][Flink学习（二）：实验一数据清洗][Flink 1](代码不完整，涉及到了elasticsearch)

\[12\][网站日志实时分析之Flink处理实时热门和PVUV统计][Flink_PVUV](缺数据)

\[13\][Is there a way to delete all the data from a topic or delete the topic before every run?][Is there a way to delete all the data from a topic or delete the topic before every run]

\[14\][redis设置密码][redis]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FwcGxleXVjaGk_size_16_color_FFFFFF_t_70]: /images/20221119/fcf3a225e9d9497a8c15138e31b08f69.png
[https_gitee.com_appleyuchi_Flink_Code_tree_master_flink_FlinkProj]: https://gitee.com/appleyuchi/Flink_Code/tree/master/flink%E6%B8%85%E6%B4%97%E6%95%B0%E6%8D%AE%E6%A1%88%E4%BE%8B/FlinkProj
[19_Flink]: https://blog.csdn.net/hongzhen91/article/details/90549079
[Flink_Kafka_MySQL]: https://blog.csdn.net/gavin_chun/article/details/85132425
[Flink]: https://www.aboutyun.com/thread-29451-1-1.html
[Flink_PV]: https://www.jianshu.com/p/52787491ea23?utm_campaign=hugo
[Flink_SLS_ACU_PCU]: https://blog.csdn.net/xiaoxiongaa0/article/details/99648125
[Flink_20_- _ETL_1]: https://blog.csdn.net/u012292754/article/details/108838196
[Flink_21_- _ETL_2]: https://blog.csdn.net/u012292754/article/details/108848949
[Flink_22_- _ETL_3]: https://blog.csdn.net/u012292754/article/details/108853574
[Flink _ Checkpoint]: https://www.jianshu.com/p/165a1bf33e4a
[Flink 1]: https://blog.csdn.net/coderpai/article/details/104898891
[Flink_PVUV]: https://blog.csdn.net/u013411339/article/details/107925298
[Is there a way to delete all the data from a topic or delete the topic before every run]: https://stackoverflow.com/questions/17730905/is-there-a-way-to-delete-all-the-data-from-a-topic-or-delete-the-topic-before-ev
[redis]: https://yuchi.blog.csdn.net/article/details/112426207