跳到主要内容
版本：Next

DataHub

DataHub Sink 连接器

描述​

DataHub Sink 用于将 SeaTunnel 数据写入阿里云 DataHub。

该连接器支持单表写入和多表写入。多表写入时，可以在 topic 中使用 ${table} 这类占位符，将不同输入表的数据写入不同的 DataHub Topic。

主要特性​

使用前准备​

运行 SeaTunnel 任务前，请先创建 DataHub 项目和 Topic。DataHub Topic 的结构中需要包含和上游 SeaTunnel schema 同名的字段，因为该 Sink 会按字段名写入数据。

选项​

名称类型必填默认值描述
endpointstring是-DataHub 服务地址。
accessIdstring是-访问 DataHub 使用的阿里云 Access ID。
accessKeystring是-访问 DataHub 使用的阿里云 Access Key。
projectstring是-DataHub 项目名称。
topicstring是-DataHub Topic 名称，多表写入时支持占位符。
timeoutint否3000客户端连接最大超时时间，单位为毫秒。
retryTimesint否3写入记录失败时的最大重试次数。
common-optionsconfig否-Sink 插件通用参数。

endpoint [string]​

DataHub 服务地址，通常以 http 或 https 开头。

accessId [string]​

访问 DataHub 使用的阿里云 Access ID。

accessKey [string]​

访问 DataHub 使用的阿里云 Access Key。

project [string]​

DataHub 项目名称。

topic [string]​

DataHub Topic 名称。多表写入时可以使用占位符，例如 ${table}。 ${table_name} 仅作为已废弃的兼容别名保留，新任务建议使用 ${table}。

SeaTunnel 字段名需要和 DataHub Topic 中的字段名一致，因为 sink 会按照 Topic 结构里的字段名写入数据。

timeout [int]​

客户端连接最大超时时间，单位为毫秒。

retryTimes [int]​

写入记录失败时的最大重试次数。

通用选项​

Sink 插件通用参数，请参考 Sink Common Options。 多表写入时，可以配合通用参数中的 multi_table_sink_replica 使用。

示例​

单表写入单个 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
  }
}

多表写入匹配的 Topic​

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

变更日志​

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