00:00
那么接下来我们可以看一个快速开始的案例,这个是官网给的一个例子。嗯,在讲这个案例之前呢,我想有必要说一下我们现在这个跑的一个集群环境,我们现在跑的是弗林克斯丹模式。那么在这个哈杜102上呢,是我们的manager。我就打个job m吧,完了之后呢,在这个103和104上呢,是我们的TM manager。就说我们有一个GM,一个TM完了之后两个TM呢,各有一个槽位。啊,因为我们只需要跑一些很简单的一个案例。啊,这是我们整个集群的情况。那么接下来呢?我就直接复制这个脚本。那就直接选中复制。完了之后到我们的这个。服务器上。我们还是CD到底下去。然后我们这个案例呢,就是把这个。
01:01
我们自己编写的这些脚本呢,配置文件啊,全部放在这个com这个这个目的下。然后直接一粘。那么接下来呢,给大家解释一下这个配置文件都干了什么事。嗯,首先呢,我们要看哪里呢?看这个英文这个块,这个块呢,实际上是给我们的啊,整个fli job指定了一个环境的参数,嗯,这里的话就是设定了我们执行的一个变程度啊,最大是一,然后呢,呃,其实我们在这里面还可以声明一些跟检查点有关的东西。啊,但是我们这个地方给束缚掉了,也就是说它在真正执行的时候不会有效。然后我们还可以看第二个块。这个地方格式有问题。嗯,实际上呢,我们可以看到。
02:00
就是首先呢,是你要有外面这个大块。这个呢说明了一个什么呢?这个实际上是一个区域完了之后,你声明完这个区域之后。你在这里面写的是一个个具体的插件。呃,那么写在source区域里面呢?就只能是source插件,写在transform区域里面的就只能是transform转换插件,写在think区域里面的同理也只能是S插件。然后每个插件呢,我们说都有一些我们可以指定的参数,你像这个哈豆102。是吧,这个host哪个机器上我们可以看到,我们这里面呢,是声明了一个socket stream,也就是说通过网络接口来获取数据,然后我们可以指定它跑在哪个host上。啊,完事呢,有两个比较重要的字段,一个是这个result table name,也就是最后这个数据呢,它要输出到哪张表里面去,另外一个就是final name,这个表的这个字段是什么?
03:04
呃,后面呢,我们会带着大家去详细的看一下这个塔诺的文档,到时候会有更详细的一个说明。呃,后面的呢。我们可以看到这里。它也是一个块完了之后呢,里面呢,我们写到了两个插件。嗯,这个地方呢,格式还是有些问题。嗯,这个地方呢,需要提前说明的一件事情啊。是这个split和circle。因为我们知道,就是之前我们说过,在flink里面目前只支持这两个转换插件。那么这个死不利的插件呢?需要说明的是,我们在SP这个地方指定了一个分隔符。然后呢,我们就是说意思呢,就是说把这个流里面的数据。然后按照这个井号分隔,然后分隔完的数据呢,分成两个字段。一个是name,一个是age。
04:00
但是需要注意的一点就是,这个split这个插件它不会立刻作用在数据上。而是什么呢?而是它会等到后面这个circleq语句,就是circleq插件里面的SQ语句。调用了这个split这个函数,然后才会。真正的作用到数据上。这个后面我们带大家看split源码的时候,也会着重说明这个事情。然后最后呢,就是说我们把这个。Think里面声明了一个console。也就是说输出到控制台。嗯,那么这个flink的任务提交上去之后呢,它这个标准输出实际上是在那个flink的web UI上可以看到的。啊,那么这就是配置文件的一个基本情况。我们后面的话去真正跑一下,看看什么效果。哦,那个配置文件呢,我们就基本上讲完了啊,最后还需要忘了说一点就是。
05:02
呃,这个source呢,它实际上每个插件呢,它是有默认值的,呃,具体的默认值的情况呢,也需要去文档里面看,实际上这个socket stream,举个例子哈,这个socket stream它的默认值呢,实际上是host是local host也是本地。然后完了之后pot。它默认监听的端口就是四个九啊,就是它的一些默认值的情况,到时候啊,你需要去看这个文档里面的具体情况。嗯,这样的话就是我们直接保存退出。那么清了一下屏,我们现在呢。呃,要先开启flink的集群。啊,看一下是不是都启动了。然后我们现在需要去开一个net cat的服务,因为我们是定义的是socket stream。
06:02
我们需要一个服务,去网站里面发数据。啊,这样的话就是那CA已经在监听9999端口。啊,这边的话呢,我们直接去提交我们那个脚本。看一下它的效果如何。你还是先4D过去。然后呢?通过这个这个参数去指定我们刚才的那个。配置。然后回车。啊,等到你看到这个job已经被提交的时候呢。啊,其实它就是已经跑在我们的弗利集群上了。现在呢,我们在这边发一点数据过去看看那边的效果。
07:01
要李四用井号分格,然后比如说他是23岁。然后张三。21岁。然后王五,我们这次用逗号风格试一下。22岁。然后呢?是赵六。我们给两个井号看一下。那么现在的话,我们就是去。Flink的VB,望上看一眼,这个端口是8081。那我们看到两个槽位呢,已经被用到了一个,然后running jobs有一个正在跑的任务。啊,我们直接去看一下它的一个控制台的输出。我们看这个task manager有一个log。然后呢,这里面是看不见的,在这个std里面,就是标准输出里面,然后我们可以看到,就是我们刚才呢,在这个。
08:03
运历行里面去发的这个数据已经跑到这个里面的任务里面来了,然后我们可以看到呢,一个是李四23岁,他被分成了两个字段,一个是李四,然后一个是23。然后还有一个就是张三,21岁。我们可以看到他也被正确的分隔了,然后王五呢,这是用逗号分隔了22岁,哎,他是什么呢?他是王五,22成立一个字段。然后后面呢,有一个字段跟的是,那其实这是一个字段哈,因为你可以看到就是22之前的这个逗号,它没有空格,然后逗号跟着后面跟着空格的是一个字段。然后这六呢,我们这里给了两个井号,实际上呢,它只会以第一个井号来作为分隔。之后呢?后面的21和22,它实际上是作为一个字段去存储的。接着呢,我们再看一下它跟我们这个SQL语句之间的关系。
09:03
哎,把它定义一下,我们去这个word文档里面看。啊,也可以看到,就是其实在这个地方,我们是查询了三个字段。一个是info,也就是原始数据完了之后split info呢,实际上是把原始数据再拆成两个字段。跟我们这个输出的对应关系,可以看到info的话,这个地方是我们第一列输出的原始数据,然后拆分的split info。这个实际上是我们后面调用的自定义函数,也就是。这个SP的插件调用了它来对这个原始数据做了一个切分,那么切分的字段名呢?实际上就是name和age。那么说到这里有一个问题大家有没有注意到,就是这个in for,它是哪来的呢?那当然是从这个fake这个表里面来的呢?那么这个表从哪来的呢?实际上是我们的source插件指定的。这个地方你要注意,就是我们在声明STEM这个插件的时候,Socket stream这个插件的时候。
10:04
我们声明了一个,也就是结果表。所以说我们从net cat发来的数据会插入到一个名为F的这个表里面。完了之后呢,插进来的数据呢,还需要有一个字段,这个表的字段呢,就是我们这里指定的file name,也就是in这个字段。因此呢,你后面的插件呢,实际上也就是我们说的,我们说的这个SQL插件,实际上呢,你在查询。这个source插件给出来的表。而且从它里面拿到了in字段。然后你再通过这个SPSP的这个自定义函数对info里面的数据进行了一个切分,啊,最终得到的是我们的这个输出结果。这也就是我们刚才写的这个配置文件,它的一个工作流程。到这里呢,我们的第一个事例就算是结束了。
我来说两句