版本：Next

Console

Console 数据接收器

支持连接器版本

所有版本

支持的引擎

Spark
Flink
SeaTunnel Zeta

描述

接收 Source 端传入的数据，并打印到 SeaTunnel 任务日志中。Console 支持批处理和流处理，主要用于调试、本地验证和示例任务，不适合作为生产环境的持久化存储。

对于每一条非空数据，Console 会打印子任务编号、行编号、表 ID、行类型和字段值。数组、Map、嵌套行等复杂类型会先转换为易读的字符串再打印。

主要特性

Console 可以接收多个上游表的数据，也可以处理 Schema 变更事件并用于日志展示。由于它写入的是日志，不向外部系统持久化数据，因此不提供精确一次写入语义，也不属于真正的 CDC 写入目标。

接收器选项

名称	类型	是否必须	默认值	描述
common-options		否	-	Sink 插件通用参数，详情请参考 Sink 常用选项。
log.print.data	boolean	否	true	是否将行数据打印到任务日志。如果只想保留 Console 节点但不打印每行数据，可以设置为 `false`。
log.print.delay.ms	int	否	0	每处理一行后的等待时间，单位为毫秒。调试时可以用它放慢打印速度。
multi_table_sink_replica	int	否	1	多表写入时每张表对应的 Sink Writer 副本数。

输出格式

Console 会在 Writer 启动时先打印行类型，然后按如下格式打印每行数据：

subtaskIndex=<子任务编号>  rowIndex=<行编号>:  SeaTunnelRow#tableId=<表 ID> SeaTunnelRow#kind=<行类型> : <字段1>, <字段2>, ...

subtaskIndex 表示打印该行的 Sink 子任务。
rowIndex 是每个 Sink Writer 内部独立递增的行编号。
tableId 表示上游表标识。单表任务通常显示为 -1。
row-kind 表示行变更类型，例如 INSERT、UPDATE_BEFORE、UPDATE_AFTER 或 DELETE。

任务示例

简单示例

下面的示例生成 3 行数据，并打印到任务日志。

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    plugin_output = "fake"
    row.num = 3
    schema = {
      fields {
        name = "string"
        age = "int"
      }
    }
  }
}

sink {
  Console {
    plugin_input = "fake"
    log.print.data = true
    log.print.delay.ms = 0
  }
}

多数据源示例

通过 plugin_input 可以把不同上游数据分别写入不同的 Console Sink。

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    plugin_output = "fake1"
    row.num = 3
    schema = {
      fields {
        id = "int"
        name = "string"
        age = "int"
        sex = "string"
      }
    }
  }
   FakeSource {
    plugin_output = "fake2"
    row.num = 3
    schema = {
      fields {
        name = "string"
        age = "int"
      }
    }
  }
}

sink {
  Console {
    plugin_input = "fake1"
  }
  Console {
    plugin_input = "fake2"
  }
}

多表输入示例

当上游 Source 产生多张表时，Console 可以在一个 Sink 中打印这些表的数据。日志中的表 ID 可以帮助区分每行数据来自哪张表。

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    plugin_output = "fake"
    tables_configs = [
      {
        row.num = 2
        schema = {
          table = "test.table1"
          columns = [
            { name = id, type = bigint }
            { name = name, type = string }
          ]
        }
      },
      {
        row.num = 2
        schema = {
          table = "test.table2"
          columns = [
            { name = id, type = bigint }
            { name = age, type = int }
          ]
        }
      }
    ]
  }
}

sink {
  Console {
    multi_table_sink_replica = 1
  }
}

控制台示例数据

控制台打印的输出:

2022-12-19 11:01:45,417 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - output rowType: name<STRING>, age<INT>
2022-12-19 11:01:46,489 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=1: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: CpiOd, 8520946
2022-12-19 11:01:46,490 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=2: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: eQqTs, 1256802974
2022-12-19 11:01:46,490 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=3: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: UsRgO, 2053193072
2022-12-19 11:01:46,490 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=4: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: jDQJj, 1993016602
2022-12-19 11:01:46,490 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=5: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: rqdKp, 1392682764
2022-12-19 11:01:46,490 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=6: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: wCoWN, 986999925
2022-12-19 11:01:46,490 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=7: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: qomTU, 72775247
2022-12-19 11:01:46,490 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=8: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: jcqXR, 1074529204
2022-12-19 11:01:46,490 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=9: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: AkWIO, 1961723427
2022-12-19 11:01:46,490 INFO  org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter - subtaskIndex=0 rowIndex=10: SeaTunnelRow#tableId=-1 SeaTunnelRow#kind=INSERT: hBoib, 929089763

变更日志

Change Log

Change	Commit	Version
[improve] console sink options (#8743)	https://github.com/apache/seatunnel/commit/c439b99f19	2.3.10
[Improve] restruct connector common options (#8634)	https://github.com/apache/seatunnel/commit/f3499a6eeb	2.3.10
[Improve][dist]add UT class name check (#8182)	https://github.com/apache/seatunnel/commit/9cf4192fe4	2.3.9
[Feature][Core] Support cdc task ddl restore for zeta (#7463)	https://github.com/apache/seatunnel/commit/8e322281ed	2.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
[Feature][Core] Add event notify for all connector (#7501)	https://github.com/apache/seatunnel/commit/d71337b0e9	2.3.8
[Improve][Connector] Add multi-table sink option check (#7360)	https://github.com/apache/seatunnel/commit/2489f6446b	2.3.7
Update ConsoleSinkFactory.java (#7350)	https://github.com/apache/seatunnel/commit/921662722f	2.3.7
[Feature][Core] Support using upstream table placeholders in sink options and auto replacement (#7131)	https://github.com/apache/seatunnel/commit/c4ca74122c	2.3.6
[Feature][Core] Support event listener for job (#6419)	https://github.com/apache/seatunnel/commit/831d0022eb	2.3.5
[Improve] Remove use `SeaTunnelSink::getConsumedType` method and mark it as deprecated (#5755)	https://github.com/apache/seatunnel/commit/8de7408100	2.3.4
[Improve] Add default implement for `SeaTunnelSink::setTypeInfo` (#5682)	https://github.com/apache/seatunnel/commit/86cba87450	2.3.4
[Feature] Support multi-table sink (#5620)	https://github.com/apache/seatunnel/commit/81ac173189	2.3.4
[Improve] Refactor CatalogTable and add `SeaTunnelSource::getProducedCatalogTables` (#5562)	https://github.com/apache/seatunnel/commit/41173357f8	2.3.4
[Feature][api env] Add job-level configuration for checkpoint timeout. (#5222)	https://github.com/apache/seatunnel/commit/3c13275ed9	2.3.4
[Improve][CheckStyle] Remove useless 'SuppressWarnings' annotation of checkstyle. (#5260)	https://github.com/apache/seatunnel/commit/51c0d709ba	2.3.4
[Feature][CDC][Zeta] Support schema evolution framework(DDL) (#5125)	https://github.com/apache/seatunnel/commit/4f89c1d272	2.3.3
Merge branch 'dev' into merge/cdc	https://github.com/apache/seatunnel/commit/4324ee1912	2.3.1
[Improve][Project] Code format with spotless plugin.	https://github.com/apache/seatunnel/commit/423b583038	2.3.1
[hotfix][zeta] fix zeta multi-table parser error (#4193)	https://github.com/apache/seatunnel/commit/98f2ad0c19	2.3.1
[Feature][Zeta] Support shuffle multiple rows by tableId (#4147)	https://github.com/apache/seatunnel/commit/8348f1a108	2.3.1
[Improve][build] Give the maven module a human readable name (#4114)	https://github.com/apache/seatunnel/commit/d7cd601051	2.3.1
[Improve][Project] Code format with spotless plugin. (#4101)	https://github.com/apache/seatunnel/commit/a2ab166561	2.3.1
[Improve][Connector-V2]console sink output content to slf4j log (#3745)	https://github.com/apache/seatunnel/commit/82a5c852d8	2.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)	https://github.com/apache/seatunnel/commit/226dc6a119	2.3.0
[improve][connector] The Factory#factoryIdentifier must be consistent with PluginIdentifierInterface#getPluginName (#3328)	https://github.com/apache/seatunnel/commit/d9519d696a	2.3.0
[Improve][Connector-V2][Console] Add Console option rule (#3322)	https://github.com/apache/seatunnel/commit/efb4711600	2.3.0
[Improve][connector][console] print subtask index (#3000)	https://github.com/apache/seatunnel/commit/de345783d9	2.3.0-beta
[Bug][Connector-V2] Fix the bug that can not print SeaTunnelRow correctly (#2749)	https://github.com/apache/seatunnel/commit/9365d35200	2.2.0-beta
[Feature][Connector-V2] Add iceberg source connector (#2615)	https://github.com/apache/seatunnel/commit/ffc6088a79	2.2.0-beta
[Bug][ConsoleSinkV2]fix fieldToString StackOverflow and add Unit-Test (#2545)	https://github.com/apache/seatunnel/commit/6f87094569	2.2.0-beta
[Improve][Console] improve console to printf schema and deepToString fields (#2517)	https://github.com/apache/seatunnel/commit/963387d375	2.2.0-beta
[api-draft][Optimize] Optimize module name (#2062)	https://github.com/apache/seatunnel/commit/f79e3112b1	2.2.0-beta

Console

支持连接器版本​

支持的引擎​

描述​

主要特性​

接收器选项​

输出格式​

任务示例​

简单示例​

多数据源示例​

多表输入示例​

控制台示例数据​

变更日志​