尚硅谷基于腾讯云EMR搭建离线数据仓库（2023版）/视频/032-腾讯云EMR-离线数仓-FlinkCDC完整代码说明.mp4原创

好，那我们测试完这个简单的弗林格CDC代码之后呢，我们看一下这个复杂的代码啊，那首先第一步还是准备这个执行环境，第二个呢，是关于checkpoint检查点的设置啊，呃，第三部分呢，就是一个参数，首先呢有一个HDFS的一个地址啊，所以刚才呢，我们是看到中间有找这个hdfs UI uii的一个地方啊，等会儿呢，我们会用到，以及我们的日期，呃，那因为从这个blog当中，日期呢，它是以当前时间来的，而我们要的应该是数据的我们自己造的这个时间，比如说我们刚才造的3月22号的数据，那我们希望我们的数据里边就应该是3月22号，这样会更好一点，对吧？呃，然后呢，是否是第一次导入啊，如果不是第一次的话呢，我们后续会有处理啊好，那接下来设置状态后端之后这一块东西呢，还是一样的，那我们把这个IP啊，就是你可以。

批，反正同时我们要打包的集群运行，呃，这个呢，写哈101也可以啊，然后J末所有的数据，然后是root啊六个零，然后接下来看，如果它不等于呢，或者说呢，呃，同时呢，它等于一的话，那表示我们是第一次导，第一次导的话，那我们要初始化，也就是说全量数据我们都要用，如果不是，那我们就用latest。OK吧，好。呃，那接下来呢，就是读取数据啊，转换数据结构，因为刚才我们看到这个数据啊，它是有这种接森套接森啊像不太好，而且呢，同时这里边有很多数据呢，对于我们来说是没有意义的啊，所以呢，我们就把这里边数据做了一个更改，做了一个调整，同时对这个时间，因为它默认的这个时间呢，用的是当前时间，对吧？因为我们可能自己生成数据嘛，可能生成的是以前的时间，所以我们把刚才这个Mo data把它引进来作为我们的时间，最终呢，封装成一个杰森格式，诶那就是杰森里边就是table，呃，Tap类型，对吧？然后时间，还有这个数据本身，以及我们的old老的数据啊，这个old呢，就在于我如果是做了一个更新操作，它是不是有before，大家还记得吧，对吧，包括这个删除，它也是有before数据，没有after，这个就相当于after，这个相当于数据OK吧，那之后呢，呃，利用我们之前想要的工具类，把这个数据呢，写到卡卡的主题里边。

啊，那在有了刚才我们简单代码的基础，这个呢也就并不难了，OK吧，啊，我们先开到这儿啊，接下来呢，我们就要去打包到集群上去运行，看能不能采集数据到我们的卡夫卡当中。