38_尚硅谷_Flume高级_自定义Sink（需求分析）

好，那我们继续，那接下来呢，我们要聊一聊这个自定义的一个S是因为。需求对吧，啊，他那个官方提供的不够用了，那同样的道理了。对吧，就还是刚才那个例子，你们那个公司的系统文件系统自己开发的，那现在我是不是想通过诶采集本地编辑系统传到我的一个系统里边呢？是不是可能有这种需求啊，对吧？那传过去是不是就要用到think啊，往那个自己的文件系统里面去写，就要用到think啊，在think里面我们看一下我们的需求啊，在这里面呢，需要不断的去轮寻China里边的事件，且批量的移除它们。啊，第一个，呃，这个呢，移除不是我们干的，是China那边干的事，假如说事务你提交了，他正常的成功了，他就会去干什么移除，因为当前事物完成了，就说明你这个think已经把数据读走了，写出去了吧，哎，也就是这一套过程都结束了，那我们就可以干什么事。

再掺到里边，这个数据可以不要了吧，但是在这个事物完成之前，不好意思，你能删掉吗。不能，因为随时可能会干什么回滚对吧？啊，随时可能会回滚是这个意思啊，不断的轮寻且批量的，这个呢，然后说并将这些事件写入到那个存储系统，所引系统，或者发送到另一个A的，这个就下有的一个问题了吧，啊随便你，因为你可以配以很多SFS啊，本地文件系统，或者说是阿弗罗的一个S，这个都可以啊，都可以传到端口里边啊，这个都无所谓，然后呢，这是完全失误的，这一块你要注意，等会我们在写代码的时候来想一下，刚才呢，我们所说的是。Source将数据传给China就行了，由China process这个东西跟谁进行对接？跟我们所说的China也就中间的事物进行对接的。但是现在。我们think来读取数据之后。读取数据之后，是不是你既然自定义C，那你核心的业务逻辑是不是要把这个读取到的给他干什么。

写出去。对吧，写出去，也就是说在think里面，这个事物就不是系统管的事了，是你自定义的时候要干什么要写出来的，因为逻辑操作是你做的。它不像刚才我们所说的source那边是不是有个China process这个组件帮我们做，但是我们当时在聊整个详细流程的时候，到了S这边有什么S。没有啊，你自定义的时候，这个时候呢，这个事物就我们自己来写啊，自己来写，当然呢，我们所说的自己来写一个事物，不是说你真的写一个事物的什么实现类啊等等这些东西，而是说你要调用它channel里边的一个事物。对吧，叫take事误，人家已经写好了，但是你要显示的给他写明出来，说是误或取是我，然后再干什么开启，开启处理事件，然后commit，如果失败，你要抓异常，干什么亏本，这个事情要你自己做了能听懂啊，因为这个，因为逻辑处理在你自己手手里，这边他没有一个什么single process，所以呢，要自己做这个事啊，自己做这个事OK，那下面呢，说的single是完全事务性的，对吧，在批量删除之前，每一个single用China启动一个事物。

啊，这个事物呢，都是China这边的。要不然它怎么会叫put叫take呢？你想想看，这个put跟take这个词用的是针对于谁来用呢？是不是China对吧？啊，也就是说我们刚说了China这边呢，是作为一个中间组件的啊，你无论是put还是take这个动作都是相对于China来说的吧，两个事物都是China这边的，OK，那接下来说了，批量事件，或者说单个事件一旦成功写入下一个系统，对吧，或者说是这个A着的这个六六十五进行提交。哦，提交那拆那边就知道了，那China就要干什么。删自己的啊，把这个数据删了，因为如果说再不删，下一次又删了就重复数据了，是不是事物整个完成了啊，那你就要删掉就要删掉，OK，那这个里边说了，包括有很多的一个组件，但是呢，呃，有的需求比较特殊，就不够对吧，我们需要自定义，官方呢，也提供了自定义的一个什么接口，我们来可以看一下啊，可以看一下这个呢，还是官方文档大家点开。

在develop开发的文档里面嘛，啊，自定义的什么东西，在开发的文档里面，而同样的我们把它呢，也改成什嘛1.7对吧？啊还是看1.7的，其实呢，这两个没差啊，这两个没差，我们就到这来，这是不是有自定义的组件，这有一个什么C呢，点开诶你看一下。我们结合着source那个来捋一下，第一个你既然跟框架进行对接，是不是应该继承框架内对吧，还有一个这个也有啊，之前在S里边它这个组件呢，就是用来读取这个。配置文件的，当时我们不是也是的，有默认的配置文件，还有。

不带默认值的对吧，两种啊在S里think里面呢，照样的可以配这些东西啊，照样的可以配这些东西，OK，那接下来你看它去读取什么。配置文件对吧，还是同样的get c，然后P跟一个什么默认值，而且大家现在已经清楚了这个P。将来要放到哪的？跟配置文件里边对应就行了，至于它这个K怎么取。是不是无所谓啊，啊，随便你想怎么去怎么去啊，无所谓好那接下来这个地方呢，它同样的叫什么stop stop啊等会呢，我们也看不到这个内容，是不是当时source就没看到这个内容啊，对吧？啊这个呢，我们就不用管啊不用管OK，最核心的就在哪。Process啊，它也有一个process，那这边还是同样的要返回值是一个。什么状态了？里面有两个状态还记得吗？Ready和大家告诉我，如果说我们数据都正常的接收完了，处理完了。

事物提交了，那我就返回什么ready对吧，接着接收下一批数据了吧，哎，如果说转移成那就干什么。大家告诉，那我们看一下是不是这个逻辑，第一它首先定义了一个等于呢，对吧，因为下面呢，重新赋值要根据两种情况进行复值，刚才我说了成功不失败，好接下来他获取了一下什么。先获取的China。啊，先get China，然后通过China去拿什么事五啊，我们之前说过这个事物是不是China里面的对吧？啊做的事物这个事物OK，那然后就开启当前我们的。15啊，15开启，开启之后注意注意这块一定要什么。传开防止发生异常对吧，等于说发生异常是失败了嘛，失败了那你看一下这个里边它的概率是。叫take了吧，之前前一个事物里面，我们是不是看到food对吧，Food事物，Take事物啊，这个呢，是由你从China里边是不是调用China。

去take呀，对吧，是不是从China里边去获取。数据，而且获取出来的，你看返回值是一个什么。Even对吧，Even OK，那这块呢，他打了两个注释。就说这块是你要自己写的一个内容了，对吧，你看一下叫send the event to。外部的一个。存储或者说配置信息，还有。存储一些数据对吧？啊就无所谓的，就是说这个是你的逻辑随便，那么我们写的呢，就打一张控制台，因为这样比较方便，那这个就看你对接的下游是什么内容，假如说你要想存在文件里边，同样的我可以开一个什么。输出流写到文件里边吧，啊是一样的道理，你写到MY搜狗那这块呢，获取JDBC是不是写到MY搜狗里边对吧？啊，你写到的就是你业务问题了啊来核心的逻辑，人家只是说核心的你要做这些步骤，好你这个写完之后，接下来他是干什么。

提交，提交完成之后，他要返回。Ready对吧，我们之前说了，那提交完成之后要返回ready啊，那这个里面呢，如果说抓到异常，你看他做什么事，是不是做回滚呢啊做回滚啊，当然你可以打印一些什么事情啊之类的啊，无所谓，然后这个状态就要改成大家告诉啊，因为他有问题了啊，他有问题了，所以呢，要稍微的去干啥推避一下。对吧，啊，C没处理的过程当中抓到异常了嘛，就我们当时C组里边是不是看到这个拜高退避啊，对吧？啊，他要退避啊之后呢，就说这个，至于这块无所谓，就是发送这个什么异常啊，错误信息啊，这个无所谓，最后最重要的是干什么。返回我们的一个什么状态信息，对吧？啊，返回我们那个状态信息是这个意思啊，返回我们状态信息，那最终这个方法的返回值就有了啊，其实这个核心逻辑如果说去掉事物这块内容的话。去掉事物这块内容的话，是不是跟source是一样的，对吧？Source里边来看一下是不是接收数据，只不过说有一点区别，Source那个地方是从外部文件系统或者说MY等等其他这个地方来干什么接收数据，而think这边呢？

一定是从China里面来获取数据吧，啊，也就是获取数据的地方不一样，然后处理就是你公司的业务逻辑。就看你公司当中想怎么对这个事情进行处理，对吧？啊是这个点啊，之后呢，由于这里边多了一个事物，所以呢，他写的就会复杂一点，而且呢，其实我们所说的这个事物呢，你只要按照这个步骤去写一下事物还是China里边已经定义好的了啊，China那边直接从China里边拿到啊，China里面拿到OK，这个是我们所说的一个点啊，那之后呢，我们来看一下我们所要做的事情啊，所要做的事情正好在这里边讲一下那个logo附近啊，之前一直说，呃，有人说log附近比较多见到这个，那log附界这个东西怎么打印数据的，哎，等会我们来看一下，我们来看一下好，第一部分要写代码。

写代码的时候还同样的，第一个MY这类要继承。抽象类实现一个接口对吧？啊，这里面呢，就有相应的一个方法，第一个。读取配置文件当中的配置信息，通过keep这种方式来获取的，对吧？context.get c get in什么get double无所谓对吧，Get什么东西，而且呢，它有两个，就是重载的方法，是不是有两个，一个只传K，一个除了传K还要传一个默认值defat value，对吧？啊，这个内容获取配置信息的在里面呢，我们都见过呢，也一样，接下来最核心的就干什么。Process啊，这个方法，这个方法呢，从China里面获取数据，同时要加上我们所对应的事物，然后呢，给它添加前后缀，这个添不添加都无所谓了。啊，其实添加前后对呢，也是同样的，在这里边去看一下，一个默认配置，一个不带默认配置两个参数两种对吧？啊其实呢啊都一样，跟那个sources那边写的都一样，等会我们加一下，最终呢，我们是要将它写入日志里边，注意这块不是说直接反映到后面来。

不是用C。对吧，Set out不是不是用，而是用什么写到日志里边，用log这种方式。啊，你可以把你输入过来的数据定义成音级别，也可以定义成error级别，或者第八个级别等等都可以啊等我们看到这个捞O件人家怎么用的，因为捞O键大家现在看到的比较多吧，什么框架里边都有捞O件啊，这个东西都是怎么用的对吧？啊OK，那之后呢，打包对吧，打包扔到那个立下啊里下，然后编写我们要的那个。配置文件对吧？啊配置文件之后呢，就是可以做测试启动，启动之后呢，做测试用控制台输入啊，控制台输入啊，既然控制台的输入的话，我们还是用那个。Source是不是随便选一个对吧？啊，我们还是选用，因为它测试比较方便是不是啊，OK，那接下来是经过China还是用memory，最后呢，经过我们自定义的这个sign诶。

我们输入的是硅谷到China里面还是硅谷，最后由于我们添加了前后缀，前面加一个hello，后面加个什么hello，你可以不加hello，是不是可以在配置文件里面去配啊，这个就是细节问题了啊，OK，这个是我们所说的整个自定义的它的一个需求啊，需求和它的一个整个的一个分析，我们应该如何去自定义这个性啊。