Skip to main content
Version: Next

Console

Console sink connector

Support Connector Version​

  • All versions

Support Those Engines​

Spark
Flink
SeaTunnel Zeta

Description​

Used to print upstream rows to the SeaTunnel task log. Console supports both batch and streaming jobs. It is mainly used for debugging, local verification, and examples, not for durable production storage.

For each non-empty row, Console prints the subtask index, row index, table id, row kind, and field values. Complex values such as arrays, maps, and nested rows are converted to readable strings before printing.

Key Features​

Console can receive rows from multiple upstream tables and can apply schema change events for display. Because it writes to logs, it does not provide exactly-once delivery or CDC writes to an external system.

Options​

NameTypeRequiredDefaultDescription
common-optionsNo-Sink plugin common parameters. See Sink Common Options for details.
log.print.databooleanNotrueWhether to print row data to the task log. Set it to false when you only want to keep the sink in the job graph without printing every row.
log.print.delay.msintNo0Delay in milliseconds after each row is processed. It can slow down printing during debugging.
multi_table_sink_replicaintNo1Writer replica count for each table in a multi-table sink job.

Output Format​

Console prints the row type once when the writer starts, and then prints each row in this format:

subtaskIndex=<subtask>  rowIndex=<index>:  SeaTunnelRow#tableId=<table-id> SeaTunnelRow#kind=<row-kind> : <field1>, <field2>, ...
  • subtaskIndex is the sink task that printed the row.
  • rowIndex is counted inside each sink writer.
  • tableId identifies the upstream table in multi-table jobs. Single-table jobs usually print -1.
  • row-kind shows the row change type, such as INSERT, UPDATE_BEFORE, UPDATE_AFTER, or DELETE.

Task Example​

Simple​

This example generates three rows and prints them to the task log.

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
  }
}

Multiple Sources Simple​

Use plugin_input to send different upstream datasets to different Console sinks.

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"
  }
}

Multi-Table Input​

When the upstream source produces multiple tables, Console can print all of them in one sink. The table id in the log helps distinguish which table produced each row.

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
  }
}

Console Sample Data​

This is a printout from our console

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

Changelog​

Change Log
ChangeCommitVersion
[improve] console sink options (#8743)https://github.com/apache/seatunnel/commit/c439b99f192.3.10
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[Improve][dist]add UT class name check (#8182)https://github.com/apache/seatunnel/commit/9cf4192fe42.3.9
[Feature][Core] Support cdc task ddl restore for zeta (#7463)https://github.com/apache/seatunnel/commit/8e322281ed2.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Feature][Core] Add event notify for all connector (#7501)https://github.com/apache/seatunnel/commit/d71337b0e92.3.8
[Improve][Connector] Add multi-table sink option check (#7360)https://github.com/apache/seatunnel/commit/2489f6446b2.3.7
Update ConsoleSinkFactory.java (#7350)https://github.com/apache/seatunnel/commit/921662722f2.3.7
[Feature][Core] Support using upstream table placeholders in sink options and auto replacement (#7131)https://github.com/apache/seatunnel/commit/c4ca74122c2.3.6
[Feature][Core] Support event listener for job (#6419)https://github.com/apache/seatunnel/commit/831d0022eb2.3.5
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)https://github.com/apache/seatunnel/commit/8de74081002.3.4
[Improve] Add default implement for SeaTunnelSink::setTypeInfo (#5682)https://github.com/apache/seatunnel/commit/86cba874502.3.4
[Feature] Support multi-table sink (#5620)https://github.com/apache/seatunnel/commit/81ac1731892.3.4
[Improve] Refactor CatalogTable and add SeaTunnelSource::getProducedCatalogTables (#5562)https://github.com/apache/seatunnel/commit/41173357f82.3.4
[Feature][api env] Add job-level configuration for checkpoint timeout. (#5222)https://github.com/apache/seatunnel/commit/3c13275ed92.3.4
[Improve][CheckStyle] Remove useless 'SuppressWarnings' annotation of checkstyle. (#5260)https://github.com/apache/seatunnel/commit/51c0d709ba2.3.4
[Feature][CDC][Zeta] Support schema evolution framework(DDL) (#5125)https://github.com/apache/seatunnel/commit/4f89c1d2722.3.3
Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
[Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
[hotfix][zeta] fix zeta multi-table parser error (#4193)https://github.com/apache/seatunnel/commit/98f2ad0c192.3.1
[Feature][Zeta] Support shuffle multiple rows by tableId (#4147)https://github.com/apache/seatunnel/commit/8348f1a1082.3.1
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Improve][Connector-V2]console sink output content to slf4j log (#3745)https://github.com/apache/seatunnel/commit/82a5c852d82.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[improve][connector] The Factory#factoryIdentifier must be consistent with PluginIdentifierInterface#getPluginName (#3328)https://github.com/apache/seatunnel/commit/d9519d696a2.3.0
[Improve][Connector-V2][Console] Add Console option rule (#3322)https://github.com/apache/seatunnel/commit/efb47116002.3.0
[Improve][connector][console] print subtask index (#3000)https://github.com/apache/seatunnel/commit/de345783d92.3.0-beta
[Bug][Connector-V2] Fix the bug that can not print SeaTunnelRow correctly (#2749)https://github.com/apache/seatunnel/commit/9365d352002.2.0-beta
[Feature][Connector-V2] Add iceberg source connector (#2615)https://github.com/apache/seatunnel/commit/ffc6088a792.2.0-beta
[Bug][ConsoleSinkV2]fix fieldToString StackOverflow and add Unit-Test (#2545)https://github.com/apache/seatunnel/commit/6f870945692.2.0-beta
[Improve][Console] improve console to printf schema and deepToString fields (#2517)https://github.com/apache/seatunnel/commit/963387d3752.2.0-beta
[api-draft][Optimize] Optimize module name (#2062)https://github.com/apache/seatunnel/commit/f79e3112b12.2.0-beta