流计算 Oceanus 模拟上下游 Datagen Logger Blackhole

一、版本说明

二、Datagen Source

Datagen 是 Flink 自带的随机数据生成器，它可以作为数据源直接引用。详细的使用方式可参考 Flink 产品文档。

下面是 Datagen 数据源的一个示例，它生成的数据含有两个字段：第一个字段 id 是一个随机数，第二个字段 name 是一个随机字符串。

三、DDL 定义

CREATE TABLE datagen_source_table ( id INT, name STRING ) WITH ( 'connector' = 'datagen', 'rows-per-second'='1' — 每秒产生的数据条数);

四、WITH 参数

五、Logger Sink

Logger Sink 是腾讯云 Oceanus 提供的一个自定义 Logger 示例，它可以将最终的结果数据写入 TaskManager 的日志文件中，后续可以通过 Flink UI 或者控制台的日志面板查看这些日志的输出。Oceanus 平台已内置 Logger Sink。

六、DDL 定义

CREATE TABLE logger_sink_table ( id INT, name STRING ) WITH ( 'connector' = 'logger', 'print-identifier' = 'DebugData');

七、WITH 参数

八、监控指标说明

Oceanus 为 Logger Sink 增加了很多实用的统计指标。单击 Flink UI 的运行图中的 Logger Sink 算子，即可搜索并查看指标：

numberOfInsertRecords：获取输出的 +I 消息数。

numberOfDeleteRecords：获取输出的 -D 消息数。

numberOfUpdateBeforeRecords：获取输出的 -U 消息数。

numberOfUpdateAfterRecords：获取输出的 +U 消息数。

九、Print Sink（不建议使用）

Flink 内置了输出到 STDOUT（标准输出）的 Print Sink，但是由于打印的格式不符合 Oceanus 日志采集器的规则，目前不能很好地展示在界面上。我们建议使用上述 Logger Sink 来代替 Print Sink。

十、DDL 定义

CREATE TABLE `print_table` ( `id` INT, `name` STRING) WITH ( 'connector' = 'print');

十一、WITH 参数