apache-kafka - 如何在 kafka sink 连接器中添加带有 kafka 消息时间戳的列
问题描述
我正在使用 properties/json 文件配置我的连接器,当它从源连接器读取消息但没有成功时,我试图添加一个包含 kafka 时间戳的时间戳列。
我尝试添加transforms
,但它始终为空,并且我的接收器连接器“大查询”返回错误
无法更新表架构
我确实将这些配置放在 bigquery 连接器属性中
transforms=InsertField
transforms.InsertField.timestamp.field=fieldtime
transforms.InsertField.type=org.apache.kafka.connect.transforms.InsertField$Value
我的源 Config Sap 连接器
{
"name": "sap",
"config": {
"connector.class": "com.sap.kafka.connect.source.hana.HANASourceConnector",
"tasks.max": "10",
"topics": "mytopic",
"connection.url": "jdbc:sap://IP:30015/",
"connection.user": "user",
"connection.password": "pass",
"group.id":"589f5ff5-1c43-46f4-bdd3-66884d61m185",
"mytopic.table.name": "\"schema\".\"mytable\""
}
}
我的接收器连接器 BigQuery
name=bigconnect
connector.class=com.wepay.kafka.connect.bigquery.BigQuerySinkConnector
tasks.max=1
sanitizeTopics=true
autoCreateTables=true
autoUpdateSchemas=true
schemaRetriever=com.wepay.kafka.connect.bigquery.schemaregistry.schemaretriever.SchemaRegistrySchemaRetriever
schemaRegistryLocation=http://localhost:8081
bufferSize=100000
maxWriteSize=10000
tableWriteWait=1000
project=kafka-test-217517
topics=mytopic
datasets=.*=sap_dataset
keyfile=/opt/bgaccess.json
transforms=InsertField
transforms.InsertField.timestamp.field=fieldtime
transforms.InsertField.type=org.apache.kafka.connect.transforms.InsertField$Value
解决方案
旧答案 我想我明白了背后的问题
首先,您不能在任何源连接器中使用转换 InsertField,因为 msg 的时间戳值是在写入主题时分配的,因此连接器不是已经知道的,
对于 JDBC 连接器,有这张票
https:/ /github.com/confluentinc/kafka-connect-jdbc/issues/311
并且在 sap 源连接器中也无法正常工作。
第二个 BigQuery 连接器有一个错误,不允许使用 InsertField 将时间戳添加到此处提到的每个表
https://github.com/wepay/kafka-connect-bigquery/issues/125#issuecomment-439102994
因此,如果您想使用 bigquery 作为输出,现在唯一的解决方案是在加载 cink 连接器之前手动编辑每个表的架构以添加列
更新 2018-12-03 始终在 SINK 连接器中添加消息时间戳的最终解决方案。假设您要将时间戳添加到接收器连接器的每个表
在您的SOURCE CONNECTOR中放置此配置
"transforms":"InsertField"
"transforms.InsertField.timestamp.field":"fieldtime",
"transforms.InsertField.type":"org.apache.kafka.connect.transforms.InsertField$Value"
这将为每个源表添加一个名为“fieldtime”的列名
在您的SINK CONNECTOR中放置这些配置
"transforms":"InsertField,DropField",
"transforms.DropField.type":"org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.DropField.blacklist":"fieldtime",
"transforms.InsertSource.timestamp.field":"kafka_timestamp",
"transforms.InsertField.timestamp.field":"fieldtime",
"transforms.InsertField.type":"org.apache.kafka.connect.transforms.InsertField$Value"
这实际上将删除列 fieldtime 并使用消息的时间戳再次添加它
此解决方案将自动添加具有正确值的列,无需任何添加操作
推荐阅读
- c# - 如何为一个类创建两个构造函数,它们都具有字符串、字符串,但它们的用途不同?
- c# - 在 c# 的当前条件下不存在更好的写法?
- apache-kafka - Kafka 自动复制到新节点
- javascript - 无论时区如何,在 JavaScript 中显示准确日期
- php - 如何通过主题覆盖视图?(yii2)
- node.js - Azure Pipeline Node.js 在不同的端口运行 localhost 服务器(express)和客户端来运行 jest 测试
- javascript - 如何在不使用 RegEX 的情况下将字符串转换为 camelCase
- python - python 请求 - 抓取 https://imgur.com
- javascript - 成功从过滤器获取结果后如何在 Laravel 8 的刀片文件中显示
- html - 是否允许 p 标签中的按钮标签?