Docs 菜单
Docs 主页
/
MongoDB Kafka Connector
/ /

所有 Sink 连接器配置属性

在此页面上

  • Overview
  • MongoDB 连接
  • MongoDB 命名空间
  • Connector 主题
  • Connector 消息处理
  • 连接器错误处理
  • 后处理器
  • Id 策略
  • 写入模型策略
  • 主题覆盖
  • 变更数据捕获
  • 时间序列

在此页面上,您可以查看 MongoDB Kafka connector 的所有可用属性。此页面重复其他接收器连接器配置属性页面的内容。

要查看所有接收器Connector配置属性页面的列表,请参阅接收器连接器配置属性页面。

使用以下配置设置指定MongoDB Kafka Sink connector如何与MongoDB cluster连接并通信。

要仅查看与配置 MongoDB 连接相关的选项,请参阅MongoDB 连接配置属性页面。

名称
说明
connection.uri
Required

Type: string

Description:
The MongoDB connection URI string to connect to your MongoDB instance or cluster.
For more information, see the Connect to MongoDB guide

重要提示: 为避免暴露connection.uri 设置中的身份验证凭证,请使用 ConfigProvider 并设立相应的配置参数。

Default: mongodb://localhost:27017
Accepted Values: A MongoDB connection URI string
server.api.version
Type: string

Description:
The Stable API version you want to use with your MongoDB server. For more information on the Stable API and versions of the server that support it, see the Stable API MongoDB server manual guide.

Default: ""
Accepted Values: An empty string or a valid Stable API version.
server.api.deprecationErrors
Type: boolean

Description:
When set to true, if the connector calls a command on your MongoDB instance that's deprecated in the declared Stable API version, it raises an exception.

You can set the API version with the server.api.version configuration option. For more information on the Stable API, see the MongoDB manual entry on the Stable API.

Default: false
Accepted Values: true or false
server.api.strict
Type: boolean

Description:
When set to true, if the connector calls a command on your MongoDB instance that's not covered in the declared Stable API version, it raises an exception.

You can set the API version with the server.api.version configuration option. For more information on the Stable API, see the MongoDB manual entry on the Stable API.

Default: false
Accepted Values: true or false

使用以下配置设置指定 MongoDB Kafka connector 将数据写入哪个 MongoDB database 和 collection。您可以使用默认的DefaultNamespaceMapper或指定自定义类。

要仅查看与指定connector写入数据的位置相关的选项,请参阅MongoDB 命名空间映射配置属性页面。

名称
说明
namespace.mapper
Type: string

Description:
The fully-qualified class name of the class that specifies which database or collection in which to sink the data. The default DefaultNamespaceMapper uses values specified in the database and collection properties.

The connector includes an alternative class for specifying the
database and collection called FieldPathNamespaceMapper. See
for more information.

Default:
com.mongodb.kafka.connect.sink.namespace.mapping.DefaultNamespaceMapper
Accepted Values: A fully qualified Java class name of a class that implements the NamespaceMapper interface.
database
Required

Type: string

Description:
The name of the MongoDB database to which the sink connector writes.

Accepted Values: A MongoDB database name
集合
Type: string

Description:
The name of the MongoDB collection to which the sink connector writes. If your sink connector follows multiple topics, this is the default collection for any writes that are not otherwise specified.
Default: The topic name.
Accepted Values: A MongoDB collection name

如果将接收器连接器配置为使用FieldPathNamespaceMapper ,则可以根据数据的字段值指定用于接收文档的数据库和集合。

要启用此映射行为,请将接收器connectornamespace.mapper配置属性设置为完全限定的类名称,如下所示:

namespace.mapper=com.mongodb.kafka.connect.sink.namespace.mapping.FieldPathNamespaceMapper

FieldPathNamespaceMapper要求您指定以下设置:

  • 将属性映射到数据库和集合的一个或两个

  • 到数据库的keyvalue映射之一

  • 到collection的keyvalue映射之一

您可以使用以下设置自定义FieldPathNamespaceMapper的行为:

名称
说明
namespace.mapper.key.database.field
Type: string

Description:
The name of the key document field that specifies the name of the database in which to write.
namespace.mapper.key.collection.field
Type: string

Description:
The name of the key document field that specifies the name of the collection in which to write.
namespace.mapper.value.database.field
Type: string

Description:
The name of the value document field that specifies the name of the database in which to write.
namespace.mapper.value.collection.field
Type: string

Description:
The name of the value document field that specifies the name of the collection in which to write.
namespace.mapper.error.if.invalid
Type: boolean

Description:
Whether to throw an exception when either the document is missing the mapped field or it has an invalid BSON type.

When set to true, the connector does not process documents missing the mapped field or that contain an invalid BSON type. The connector may halt or skip processing depending on the related error-handling configuration settings.

When set to false, if a document is missing the mapped field or if it has an invalid BSON type, the connector defaults to writing to the specified database and collection settings.

Default: false
Accepted Values: true or false

使用以下配置设置指定 MongoDB Kafka 接收器连接器应监视哪些 Kafka 主题的数据。

要仅查看与指定 Kafka 主题相关的选项,请参阅Kafka 主题属性页面。

名称
说明
topics
Required

Type: list

Description:
A list of Kafka topics that the sink connector watches.

您可以定义topicstopics.regex设置,但不能同时定义两者。

Accepted Values: A comma-separated list of valid Kafka topics
topics.regex
Required

Type: string

Description:
A regular expression that matches the Kafka topics that the sink connector watches.

示例,以下正则表达式匹配主题名称,例如“action.landing.clicks” 和“活动.支持.点击”。 它与主题名称“action.landing.views”不匹配 和“Activity.Clicks”。

topics.regex=activity\\.\\w+\\.clicks$

您可以定义topicstopics.regex设置,但不能同时定义两者。

Accepted Values: A valid regular expression pattern using java.util.regex.Pattern.

使用此页面上的设置可配置 MongoDB Kafka connector 的消息处理行为,包括:

  • 消息批量大小

  • 速率限制

  • 并行任务数

要仅查看与变更数据捕获处理程序相关的选项,请参阅connector消息处理Properties页面。

名称
说明
max.batch.size
Type: int

Description:
Maximum number of sink records to batch together for processing.

Consider the batch that contains the following records:
[ 1, 2, 3, 4, 5 ]
When set to 0, the connector performs a single bulk write for the entire batch.

When set to 1, the connector performs one bulk write for each record in the batch, for a total of five bulk writes as shown in the following example:
[1], [2], [3], [4], [5]
Default: 0
Accepted Values: An integer
bulk.write.ordered
Type: boolean

Description:
Whether the connector writes a batch of records as an ordered or unordered bulk write operation. When set to true, the default value, the connector writes a batch of records as an ordered bulk write operation.

To learn more about bulk write operations, see Bulk Write Operations.

Default: true
Accepted Values: true or false
rate.limiting.every.n
Type: int

Description:
Number of batches of records the sink connector processes in order to trigger the rate limiting timeout. A value of 0 means no rate limiting.

Default: 0
Accepted Values: An integer
rate.limiting.timeout
Type: int

Description:
How long (in milliseconds) to wait before the sink connector should resume processing after reaching the rate limiting threshold.

Default: 0
Accepted Values: An integer
tasks.max
Type: int

Description:
The maximum number of tasks to create for this connector. The connector may create fewer than the maximum tasks specified if it cannot handle the level of parallelism you specify.

IMPORTANT: If you specify a value greater than 1, the connector enables parallel processing of the tasks. If your topic has multiple partition logs, which enables the connector to read from the topic in parallel, the tasks may process the messages out of order.

Default: 1
Accepted Values: An integer

使用以下配置设置指定 MongoDB Kafka connector 如何处理错误并配置死信队列(DLQ)。

要仅查看与处理错误相关的选项,请参阅connector错误处理属性页面。

名称
说明
mongo.errors.tolerance
Type: string

Description:
Whether to continue processing messages if the connector encounters an error. Allows the connector to override the errors.tolerance Kafka cluster setting.

When set to none, the connector reports any error and blocks further processing of the rest of the messages.

When set to all, the connector ignores any problematic messages.

When set to data, the connector tolerates only data errors and fails on all other errors.

To learn more about error handling strategies, see the Handle Errors page.

This property overrides the errors.tolerance
property of the Connect Framework.

Default: Inherits the value from the errors.tolerance setting.
Accepted Values: "none" or "all"
mongo.errors.log.enable
Type: boolean

Description:
Whether the connector should write details of errors including failed operations to the log file. The connector classifies errors as "tolerated" or "not tolerated" using the errors.tolerance or mongo.errors.tolerance settings.

When set to true, the connector logs both "tolerated" and "not tolerated" errors.
When set to false, the connector logs "not tolerated" errors.

This property overrides the errors.log.enable
property of the Connect Framework.

Default: false
Accepted Values: true or false
errors.log.include.messages
Type: boolean

Description:
Whether the connector should include the invalid message when logging an error. An invalid message includes data such as record keys, values, and headers.

Default: false
Accepted Values: true or false
errors.deadletterqueue.topic.name
Type: string

Description:
Name of topic to use as the dead letter queue. If blank, the connector does not send any invalid messages to the dead letter queue.

To learn more about the dead letter queue, see the Dead Letter Queue Configuration Example.

Default: ""
Accepted Values: A valid Kafka topic name
errors.deadletterqueue.context.headers.enable
Type: boolean

Description:
Whether the connector should include context headers when it writes messages to the dead letter queue.

To learn more about the dead letter queue, see the Dead Letter Queue Configuration Example.

To learn about the exceptions the connector defines and reports through context headers, see Bulk Write Exceptions.

Default: false
Accepted Values: true or false
errors.deadletterqueue.topic.replication.factor
Type: integer

Description:
The number of nodes on which to replicate the dead letter queue topic. If you are running a single-node Kafka cluster, you must set this to 1.

To learn more about the dead letter queue, see the Dead Letter Queue Configuration Example.

Default: 3
Accepted Values: A valid number of nodes

使用以下配置设置指定 MongoDB Kafka connector 在将 Kafka 数据插入 MongoDB 之前应如何转换该数据。

要仅查看与帖子处理器相关的选项,请参阅Sink connector 帖子处理器属性页面。

名称
说明
post.processor.chain
Type: list

Description:
A list of post-processor classes the connector should apply to process the data before saving it to MongoDB.

To learn more about post-processors and see examples of
their usage, see

Default:
com.mongodb.kafka.connect.sink.processor.DocumentIdAdder
Accepted Values: A comma-separated list of fully qualified Java class names
field.renamer.mapping
Type: string

Description:
A list of field name mappings for key and value fields. Define the mappings in an inline JSON array in the following format:
[ { "oldName":"key.fieldA", "newName":"field1" }, { "oldName":"value.xyz", "newName":"abc" } ]
Default: []
Accepted Values: A valid JSON array
field.renamer.regexp
Type: string

Description:
A list of field name mappings for key and value fields using regular expressions. Define the mappings in an inline JSON array in the following format:
[ {"regexp":"^key\\\\..*my.*$", "pattern":"my", "replace":""}, {"regexp":"^value\\\\..*$", "pattern":"\\\\.", "replace":"_"} ]
Default: []
Accepted Values: A valid JSON array
key.projection.list
Type: string

Description:
A list of field names the connector should include in the key projection.

Default: ""
Accepted Values: A comma-separated list of field names
key.projection.type
Type: string

Description:
The key projection type the connector should use.

Default: none
Accepted Values: none, BlockList, or AllowList (Deprecated: blacklist, whitelist)
value.projection.list
Type: string

Description:
A list of field names the connector should include in the value projection.

Default: ""
Accepted Values: A comma-separated list of field names
value.projection.type
Type: string

Description:
The type of value projection the connector should use.

Default: none
Accepted Values: none, BlockList, or AllowList (Deprecated: blacklist, whitelist)
writemodel.strategy
Type: string

Description:
The class that specifies the WriteModelStrategy the connector should use for Bulk Writes.

To learn more about how to create your own strategy, see

Default:
com.mongodb.kafka.connect.sink.writemodel.strategy.DefaultWriteModelStrategy
Accepted Values: A fully qualified Java class name

使用以下配置设置,指定 MongoDB Kafka Sink 连接器应如何确定写入 MongoDB 的每个文档的 _id 值。

要仅查看与确定文档_id字段相关的选项,请参阅“connector ID 策略属性”页面。

名称
说明
document.id.strategy
Type: string

Description:
The class the connector should use to generate a unique _id field.

Default:
com.mongodb.kafka.connect.sink.processor.id.strategy.BsonOidStrategy
Accepted Values: An empty string or a fully qualified Java class name
document.id.strategy.overwrite.existing
Type: boolean

Description:
Whether the connector should overwrite existing values in the _id field when it applies the strategy defined by the document.id.strategy property.

Default: false
Accepted Values: true or false
document.id.strategy.uuid.format
Type: string

Description:
Whether the connector should output the UUID in the _id field in string format or in BsonBinary format.

Default: string
Accepted Values: string or binary
delete.on.null.values
Type: boolean

Description:
Whether the connector should delete documents when the key value matches a document in MongoDB and the value field is null.

This setting applies when you specify an id generation strategy that operates on the key document such as FullKeyStrategy, PartialKeyStrategy, and ProvidedInKeyStrategy.

Default: false
Accepted Values: true or false

您可以设置配置属性来指定MongoDB Kafka Sink connector如何将数据写入MongoDB 。 以下部分介绍了可以设置的配置属性,可自定义此行为。

writemodel.strategy 配置属性设置为 ,以指定接收器connector在接收接收器记录时如何写入数据。

您可以将writemodel.strategy的值设置为本页“策略”部分中描述的写入模型策略的任何完全限定类名称。 您可以通过设置以下配置来指定策略:

writemodel.strategy=<a write model strategy>

delete.writemodel.strategy 配置属性设置为 ,以指定 Sink connector在收到 tombstone 事件时如何写入数据。 逻辑删除事件是指包含键但不包含值的记录,这表示记录已删除。

您可以将delete.writemodel.strategy的值设置为本页“策略”部分中描述的写入模型策略的任何完全限定类名称。 您可以通过设置以下配置来指定策略:

delete.writemodel.strategy=<a write model strategy>

要仅查看与写入模型策略相关的选项,请参阅接收器连接器写入模型策略页面。

名称
说明
DefaultWriteModelStrategy

Description:
This strategy uses the ReplaceOneDefaultStrategy by default, and the InsertOneDefaultStrategy if you set the timeseries.timefield option.

This is the default value value for the writemodel.strategy configuration property.
InsertOneDefaultStrategy

Description:
Insert each sink record into MongoDB as a document.
To specify this strategy, set the configuration property to the following class name:
com.mongodb.kafka.connect.sink.writemodel.strategy.InsertOneDefaultStrategy
ReplaceOneDefaultStrategy

Description:
Replaces at most one document in MongoDB that matches a sink record by the _id field. If no documents match, the connector inserts the sink record as a new document.
To specify this strategy, set the configuration property to the following class name:
com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneDefaultStrategy
ReplaceOneBusinessKeyStrategy

Description:
Replaces at most one document that matches a sink record by a specified business key. If no documents match, the connector inserts the sink record as a new document.
To specify this strategy, set the configuration property to the following class name:
com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy
To see an example showing how to use this strategy, see our guide on write model strategies.
DeleteOneDefaultStrategy

Description:
Deletes at most one document that matches your sink connector's key structure by the _id field only when the document contains a null value structure.

This is the default value for the delete.writemodel.strategy configuration property.

This strategy is set as the default value of the writemodel.strategy property when you set mongodb.delete.on.null.values=true.
To specify this strategy, set the configuration property to the following class name:
com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteOneDefaultStrategy
DeleteOneBusinessKeyStrategy

Description:
Deletes at most one MongoDB document that matches a sink record by a business key.
To specify this strategy, set the configuration property to the following class name:
com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteOneBusinessKeyStrategy
To see an example showing how to use this strategy, see our guide on write model strategies.
UpdateOneDefaultStrategy

Description:
Updates at most one document in MongoDB that matches a sink record by the _id field. If no documents match, the connector inserts the sink record as a new document.
To specify this strategy, set the configuration property to the following class name:
com.mongodb.kafka.connect.sink.writemodel.strategy.UpdateOneDefaultStrategy
UpdateOneTimestampsStrategy

Description:
Add _insertedTS (inserted timestamp) and _modifiedTS (modified timestamp) fields into documents.
To specify this strategy, set the configuration property to the following class name:
com.mongodb.kafka.connect.sink.writemodel.strategy.UpdateOneTimestampsStrategy
To see an example showing how to use this strategy, see our guide on write model strategies.
UpdateOneBusinessKeyTimestampStrategy

Description:
Add _insertedTS (inserted timestamp) and _modifiedTS (modified timestamp) fields into documents that match a business key.
To specify this strategy, set the configuration property to the following class name:
com.mongodb.kafka.connect.sink.writemodel.strategy.UpdateOneBusinessKeyTimestampStrategy

使用以下 MongoDB Kafka 接收器连接器配置设置来覆盖特定主题的全局或默认属性设置。

要仅查看与覆盖主题设置相关的选项,请参阅主题覆盖属性页面。

名称
说明
topic.override.<topicName>.<propertyName>
Type: string

Description:
Specify a topic and property name to override the corresponding global or default property setting.

For example, the topic.override.foo.collection=bar setting instructs
the sink connector to store data from the foo topic in the bar
collection.

You can specify any valid configuration setting in the
<propertyName> segment on a per-topic basis except
connection.uri and topics.

Default: ""
Accepted Values: Accepted values specific to the overridden property

使用以下配置设置指定MongoDB Kafka接收器connector用于处理变更数据捕获 (CDC) 事件的类。

有关使用 Debezium 和 Qlik Replicate事件生成器的内置ChangeStreamHandler 和处理程序的示例,请参阅接收Connector变更数据捕获指南。

要仅查看与变更数据捕获处理程序相关的选项,请参阅变更数据捕获属性页面。

名称
说明
change.data.capture.handler
Type: string

Description:
The class name of the CDC handler to use for converting changes into event streams. See Available CDC Handlers for a list of CDC handlers.

Default: ""
Accepted Values: An empty string or a fully qualified Java class name

使用以下配置设置指定 MongoDB Kafka 接收器连接器应如何将数据接收到 MongoDB 时间序列集合。

要仅查看与time-series collection相关的选项,请参阅Kafka time-series 属性页面。

名称
说明
timeseries.timefield
Type: string

Description:
The name of the top-level field in the source data that contains time information that you want to associate with the new document in the time series collection.

Default: ""
Accepted Values: An empty string or the name of a field that contains a BSON DateTime value
timeseries.timefield.auto.convert.date.format
Type: string

Description:
The date format pattern the connector should use to convert the source data contained in the field specified by the timeseries.timefield setting.

The connector passes the date format pattern to the Java DateTimeFormatter.ofPattern(pattern, locale) method to perform date and time conversions on the time field.

If the date value from the source data only contains date information, the connector sets the time information to the start of the specified day. If the date value does not contain the timezone offset, the connector sets the offset to UTC.

Default:
yyyy-MM-dd[['T'][ ]][HH:mm:ss[[.][SSSSSS][SSS]][ ]VV[ ]'['VV']'][HH:mm:ss[[.][SSSSSS][SSS]][ ]X][HH:mm:ss[[.][SSSSSS][SSS]]]
Accepted Values: A valid DateTimeFormatter format
timeseries.timefield.auto.convert
Type: boolean

Description:
Whether to convert the data in the field into the BSON Date format.

When set to true, the connector uses the milliseconds after epoch and discards fractional parts if the value is a number. If the value is a string, the connector uses the setting in the following configuration to parse the date:
timeseries.timefield.auto.convert.date.format
If the connector fails to convert the value, it sends the original value to the time series collection.

Default: false
Accepted Values: true or false
timeseries.timefield.auto.convert.locale.language.tag
Type: string

Description:
Which DateTimeFormatter locale language tag to use with the date format pattern (e.g. "en-US").

To learn more about locales, see the Java SE documentation of Locale.

Default: ROOT
Accepted Values: A valid Locale language tag format
timeseries.metafield
Type: string

Description:
Which top-level field to read from the source data to describe a group of related time series documents.

IMPORTANT: This field must not be the _id field nor the field you specified in the timeseries.timefield setting.

Default: ""
Accepted Values: An empty string or the name of a field that contains any BSON type except BsonArray.
timeseries.expire.after.seconds
Type: int

Description:
The number of seconds MongoDB should wait before automatically removing the time series collection data. The connector disables timed expiry when the setting value is less than 1.

To learn more, see Set up Automatic Removal for Time Series Collections in the MongoDB manual.

Default: 0
Accepted Values: An integer
timeseries.granularity
Type: string

Description:
The expected interval between subsequent measurements of your source data.

To learn more, see Set Granularity for Time Series Data in the MongoDB manual.

Optional
Default: ""
Accepted Values: "", "seconds", "minutes", "hours"

有关如何将现有集合转换为时间序列集合的示例,请参阅如何将现有集合迁移到时间序列集合的教程。

后退

Kafka 时间序列属性