HeaderToValue

HeaderToValue 单消息转换（SMT）会从事件记录中提取指定的标头字段，然后将这些字段复制或移动到事件记录的值中。move 选项会在把这些字段作为值添加到负载中之前，先将其从标头中彻底移除。你可以配置该 SMT 以操作原始消息中的多个标头。你还可以使用点号表示法指定负载中的某个节点，将标头字段嵌套到该节点中。有关配置该 SMT 的更多信息，请参阅以下示例。

若要将事件记录中的消息标头提取到记录值中，请在连接器的 Kafka Connect 配置中配置 HeaderToValue SMT。你可以将该转换配置为移除原始标头或复制标头。要从记录中移除标头字段，请将 SMT 配置为使用 move 操作；要保留原始记录中的标头字段，请将 SMT 配置为使用 copy 操作。例如，要从事件消息中移除 event_timestamp 和 key 标头，请在连接器配置中添加以下几行：

transforms=moveHeadersToValue
transforms.moveHeadersToValue.type=io.debezium.transforms.HeaderToValue
transforms.moveHeadersToValue.headers=event_timestamp,key
transforms.moveHeadersToValue.fields=timestamp,source.id
transforms.moveHeadersToValue.operation=move

以下示例展示了事件记录在应用转换前后的标头和值。

示例 1. 应用 HeaderToValue SMT 的效果

HeaderToValue 转换处理前的事件记录

SMT 处理事件记录前的标头

{
    "header_x": 0,
    "event_timestamp": 1626102708861,
    "key": 100,
}

SMT 处理事件记录前的值

{
        "before": null,
        "after": {
            "id": 1,
            "first_name": "Anne",
            "last_name": "Kretchmar",
            "email": "[email protected]"
        },
        "source": {
            "version": "2.1.3.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "ts_us": 1559033904863091,
            "ts_ns": 1559033904863091000,
            "snapshot": true,
            "db": "postgres",
            "sequence": "[\"24023119\",\"24023128\"]"
            "schema": "public",
            "table": "customers",
            "txId": 555,
            "lsn": 24023128,
            "xmin": null
        },
        "op": "c",
        "ts_ms": 1559033904863,
        "ts_us": 1559033904863741,
        "ts_ns": 1559033904863741697
    }

经过 HeaderToValue 转换处理后的事件记录

SMT 移除指定字段后的标头

{
    "header_x": 0
}

SMT 将标头字段移入值之后的值

{
        "before": null,
        "after": {
            "id": 1,
            "first_name": "Anne",
            "last_name": "Kretchmar",
            "email": "[email protected]"
        },
        "source": {
            "version": "2.1.3.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "ts_us": 1559033904863697,
            "ts_ns": 1559033904863697000,
            "snapshot": true,
            "db": "postgres",
            "sequence": "[\"24023119\",\"24023128\"]"
            "schema": "public",
            "table": "customers",
            "txId": 555,
            "lsn": 24023128,
            "xmin": null,
            "id": 100
        },
        "op": "c",
        "ts_ms": 1559033904863,
        "ts_us": 1559033904863631,
        "ts_ns": 1559033904863631584,
        "event_timestamp": 1626102708861
    }

下表列出了可与 HeaderToValue SMT 搭配使用的配置选项。

属性	说明	类型	默认值	有效值	重要性
`headers`	以逗号分隔的记录头（header）名称列表，这些头的值将被复制或移动到记录值中。	list	无默认值	非空列表	高
`fields`	以逗号分隔的字段名称列表，其顺序须与 `headers` 配置属性中列出的头名称顺序一致。使用点号表示法可指示 SMT 将字段嵌套到消息负载的特定节点中。有关如何配置 SMT 以使用点号表示法的信息，请参阅本主题前面的示例。	list	无默认值	非空列表	高
`operation`	指定以下选项之一：`move`:: SMT 将头字段移动到事件记录的值中，并从头中移除这些字段。`copy`:: SMT 将头字段复制到事件记录的值中，并保留原始头字段。	string	无默认值	move 或 copy	高

表 1. HeaderToValue SMT 配置选项

头字段转值

HeaderToValue

评论