Skip to main content
Version: Next

DataHub

DataHub sink connector

Description​

The DataHub sink writes SeaTunnel rows to Alibaba Cloud DataHub.

The connector supports single-table writes and multi-table writes. In multi-table jobs, use placeholders such as ${table} in topic to route records from different input tables to different DataHub topics.

Key features​

Before You Start​

Create the DataHub project and topic before running the SeaTunnel job. The DataHub topic schema must contain fields with the same names as the upstream SeaTunnel schema fields, because the sink writes values by field name.

Options​

nametyperequireddefault valuedescription
endpointstringyes-DataHub service endpoint.
accessIdstringyes-Alibaba Cloud access ID used to access DataHub.
accessKeystringyes-Alibaba Cloud access key used to access DataHub.
projectstringyes-DataHub project name.
topicstringyes-DataHub topic name. Supports placeholders in multi-table jobs.
timeoutintno3000Maximum client connection timeout in milliseconds.
retryTimesintno3Maximum retry count when writing a record fails.
common-optionsconfigno-Sink plugin common options.

endpoint [string]​

The DataHub service endpoint. It usually starts with http or https.

accessId [string]​

The Alibaba Cloud access ID used to access DataHub.

accessKey [string]​

The Alibaba Cloud access key used to access DataHub.

project [string]​

The DataHub project name.

topic [string]​

The DataHub topic name. For multi-table writes, this value can contain placeholders, for example ${table}. ${table_name} is only kept as a deprecated compatibility alias; use ${table} for new jobs.

The SeaTunnel field names must match the DataHub topic fields, because the sink writes fields by name according to the topic schema.

timeout [int]​

The maximum client connection timeout in milliseconds.

retryTimes [int]​

The maximum retry count when writing a record fails.

common options​

Sink plugin common parameters, please refer to Sink Common Options for details. For multi-table writes, multi_table_sink_replica can be used with the common sink options.

Examples​

Write one table to one topic​

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

source {
  FakeSource {
    plugin_output = "fake"
    schema = {
      fields {
        name = "string"
        age = "int"
      }
    }
  }
}

sink {
  DataHub {
    endpoint = "https://datahub.example.aliyuncs.com"
    accessId = "your-access-id"
    accessKey = "your-access-key"
    project = "demo_project"
    topic = "user_topic"
    timeout = 3000
    retryTimes = 3
  }
}

Write multiple input tables to matching topics​

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

source {
  FakeSource {
    plugin_output = "fake"

    tables_configs = [
      {
        row.num = 100
        schema = {
          table = "users"
          fields {
            name = "string"
            age = "int"
          }
        }
      },
      {
        row.num = 200
        schema = {
          table = "orders"
          fields {
            order_id = "int"
            amount = "decimal(10, 2)"
          }
        }
      }
    ]
  }
}

sink {
  DataHub {
    endpoint = "https://datahub.example.aliyuncs.com"
    accessId = "your-access-id"
    accessKey = "your-access-key"
    project = "demo_project"
    topic = "${table}"
    timeout = 3000
    retryTimes = 3
  }
}

Changelog​

Change Log
ChangeCommitVersion
[Feature][Connector-V2] Make some sink parameters optional for DataHub (#9229)https://github.com/apache/seatunnel/commit/7418fae10c2.3.11
[Feature][Connector-V2] Datahub support multi-table sink (#9212)https://github.com/apache/seatunnel/commit/7027162dec2.3.11
[improve] datahub sink options (#8744)https://github.com/apache/seatunnel/commit/88f35bd7052.3.10
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)https://github.com/apache/seatunnel/commit/8de74081002.3.4
[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
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Improve][Connector-V2][DataHub] Unified exception for DataHub sink connector & change package name of DataHub (#3446)https://github.com/apache/seatunnel/commit/395635fa182.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][DataHub] Add DataHub Sink Factory (#3323)https://github.com/apache/seatunnel/commit/685978d0612.3.0
[#2606]Dependency management split (#2630)https://github.com/apache/seatunnel/commit/fc047be69b2.2.0-beta
[Feature][Connector-V2]Support datahub sink (#2558)https://github.com/apache/seatunnel/commit/43600a70492.2.0-beta