3.1 SeaTunnel的启动脚本

3.1.1 启动脚本的参数

截至目前，SeaTunnel有两个启动脚本。

提交spark任务用start-seatunnel-spark.sh。

提交flink任务则用start-seatunnel-flink.sh。

本文档主要是结合flink来使用seatunnel的，所以用start-seatunnel-flink.sh来讲解。

start-seatunnle-flink.sh可以指定3个参数

分别是：

--config 应用配置的路径

--variable 应用配置里的变量赋值

--check 检查config语法是否合法

3.1.2 --check参数

截至本文档撰写时的SeaTunnel版本v2.1.0。check功能还尚在开发中，因此--check参数是一个虚设。目前start-seatunnel-flink.sh并不能对应用配置文件的语法合法性进行检查。而且start-seatunnel-flink.sh中目前没有对--check参数的处理逻辑。

需要注意！使用过程中，如果没有使用--check参数，命令行一闪而过。那就是你的配置文件语法有问题。

3.1.3 --config参数和--variable参数

--config参数用来指定应用配置文件的路径。

--variable参数可以向配置文件传值。配置文件内是支持声明变量的。然后我们可以通过命令行给配置中的变量赋值。

变量声明语法如下。

sql {
    sql = "select * from (select info,split(info) from fake) where age > '"${age}"'"
  }

在配置文件的任何位置都可以声明变量。并用命令行参数--variable key=value的方式将变量值传进去，你也可以用它的短命令形式 -i key=value。传递参数时，key需要和配置文件中声明的变量名保持一致。

如果需要传递多个参数，那就在命令行里面传递多个-i或--variable key=value。

比如：

bin/start-seatunnel-flink.sh --config/xxx.sh -i age=18 -i sex=man

3.1.4 示例2：配置中使用变量

1）我们在example01.conf的基础上创建example02.conf。

[atguigu@hadoop102 config]$ cp example01.sh example02.sh

2）修改文件

[atguigu@hadoop102 config]$ vim example02.sh

3）给sql插件声明一个变量，红色的是我们修改的地方。最终的配置文件如下。

env {
  execution.parallelism = 1
}
 
source {
    SocketStream{
          result_table_name = "fake"
          field_name = "info"
    }
}
 
transform {
  Split{
    separator = "#"
    fields = ["name","age"]
  }
 
  sql {
sql = "select * from (select info, split(info) from fake) where age > '"${age}"'"
# 需要套一层子查询，因为where先于select,split出的字段无法用where过滤
  }
}
 
sink {
  ConsoleSink {}
}

4）开启netcat服务

[atguigu@hadoop102 ~]nc -l 9999

5）使用SeaTunnel来提交任务。-i age=18往命令行中

bin/start-seatunnel-flink.sh --config config/example01.sh -i age=18

6）接着，我们用nc发送几条数据看看效果。

7）在flink的webUI上我们看一下控制台的输出。最终发现未满18岁的李四被过滤掉了。

8）小结

通过传递变量，我们可以实现配置文件的复用。让同一份配置文件就能满足不同的业务需求。

3.1.5 给flink传递参数

在启动脚本的尾部，我们可以看到，start-seatunnel-flink.sh会执行(exec)一条命令，这个命令会使用flink的提交脚本去向集群提交一个任务。而且在调用bin/flink run的时候，还传递了PARAMS作为flink run的参数。

如下图所示，我们可知，凡是--config 和 --variable之外的命令行参数都被放到PARAMS变量中，最后相当于给flink run传递了参数。注意！命令行参数解析过程中没有涉及--check参数处理。这也是为什么说它目前不支持--check操作。

比如，我们可以在seatunnel启动脚本中，指定flink job并行度。

bin/start-seatunnel-flink.sh --config config/ -p 2\

3.2 SeaTunnel的配置文件

3.2.1 应用配置的4个基本组件

我们从SeaTunnel的app配置文件开始讲起。

一个完整的SeaTunnel配置文件应包含四个配置组件。分别是：

env{} source{} --> transform{} --> sink{}

在Source和Sink数据同构时，如果业务上也不需要对数据进行转换，那么transform中的内容可以为空。具体需根据业务情况来定。

3.2.2 env块

env块中可以直接写spark或flink支持的配置项。比如并行度，检查点间隔时间。检查点hdfs路径等。在SeaTunnel源码的ConfigKeyName类中，声明了env块中所有可用的key。如图所示：

3.2.3 SeaTunnel中的核心数据结构Row

Row是SeaTunnel中数据传递的核心数据结构。对flink来说，source插件需要给下游的转换插件返回一个DataStream<Row>，转换插件接到上游的DataStream<Row>进行处理后需要再给下游返回一个DataStream<Row>。最后Sink插件将转换插件处理好的DataStream<Row>输出到外部的数据系统。

如图所示：

因为DataStream<Row>可以很方便地和Table进行互转，所以将Row当作核心数据结构可以让转换插件同时具有使用代码（命令式）和sql（声明式）处理数据的能力。

3.2.4 source块

source块是用来声明数据源的。source块中可以声明多个连接器。比如：

# 伪代码
env {
    ...
}
 
source {
  hdfs { ... }  
  elasticsearch { ... }
  jdbc {...}
}
 
transform {
    sql {
     sql = """
        select .... from hdfs_table 
        join es_table 
        on hdfs_table.uid = es_table.uid where ..."""
    }
}
 
sink {
    elasticsearch { ... }
}

需要注意的是，所有的source插件中都可以声明result_table_name。如果你声明了result_table_name。SeaTunnel会将source插件输出的DataStream<Row>转换为Table并注册在Table环境中。当你指定了result_table_name，那么你还可以指定field_name，在注册时，给Table重设字段名(后面的案例中会讲解)。

因为不同source所需的配置并不一样，所以对source 连接器的配置最好参考官方的文档。

3.2.5 transform块

目前社区对插件做了很多规划，但是截至v2.1.0版本，可用的插件总共有两个，一个是Split，另一个是sql。

transform{}块中可以声明多个转换插件。所有的转换插件都可以使用source_table_name，和result_table_name。同样，如果我们声明了result_table_name，那么我们就能声明field_name。

我们需要着重了解一下Split插件和sql插件的实现。但在此

在SeaTunnel中，一个转换插件的实现类最重要的逻辑在下述四个方法中。

1）处理批数据，DataSet<Row>进，DataSet<Row>出

DataSet<Row> processBatch(FlinkEnvironment env, DataSet<Row> data)

2）处理流数据，DataStram<Row>进，DataStream<Row>出

DataStream<Row> processStream(FlinkEnvironment env, DataStream<Row> dataStream)

3）函数名叫注册函数。实际上，这是一个约定，它只不过是每个transform插件作用于流之后调用的一个函数。

void registerFunction(FlinkEnvironment env, DataStream<Row> datastream)

4）处理一些预备工作，通常是用来解析配置。

void prepare(FlinkEnvironment prepareEnv)

Split插件的实现

现在我们需要着重看一下Split插件的实现。

先回顾一下我们之前example01.conf中关于transform的配置。

接着我们再来看一下Split的源码实现。

我们发现Split插件并没有对数据流进行任何的处理，而是将它直接return了。反之，它向表环境中注册了一个名为split的UDF（用户自定义函数）。而且，函数名是写死的。这意味着，如果你声明了多个Split，后面的UDF还会把前面的覆盖。

这是开发时需要注意的一个点。

但是，需要注意，tranform接口其实是留给了我们直接操作数据的能力的。也就是processStream方法。那么，一个transform插件其实同时履行了process和udf的职责，这是违反单一职责原则的。那要判断一个转换插件在做什么就只能从源码和文档的方面来加以区分了。

最后需要叮嘱的是，指定soure_table_name对于sql插件的意义不大。因为sql插件可以通过from子句来决定从哪个表里抽取数据。

3.2.6 sink块

Sink块里可以声明多个sink插件，每个sink插件都可以指定source_table_name。不过因为不同Sink插件的配置差异较大，所以在实现时建议参考官方文档。

3.3 SeaTunnel的基本原理

SeaTunnel的工作原理简单明了。

1）程序会解析你的应用配置，并创建环境

2）配置里source{}，transform{}，sink{}三个块中的插件最终在程序中以List集合的方式存在。

3）由Excution对象来拼接各个插件，这涉及到选择source_table，注册result_table等流程，注册udf等流程。并最终触发执行

可以参考下图：

3.4 小结

最后我们用一张图将SeaTunnel中的重要概念串起来。

如果你愿意，依托sql插件和udf。单个配置文件也可以定义出比较复杂的工作流。但SeaTunnel的定位是一个数据集成平台。核心的功能是依托丰富的连接器进行数据同步，数据处理并不是SeaTunnel的长处。所以在SeaTunnel中定义复杂的工作流或许是一种不值得提倡的做法。

需要提醒的是，如果你不指定source_table_name，插件会使用它在配置文件上最近的上一个插件的输出作为输入。

所以，我们可以通过使用依托表名表环境来实现复杂的工作流。

也可以减少表名的使用实现简单的数据同步通道。