Kafka 文件接收器连接器 - 输出 JSON [英] Kafka File Sink Connector - output JSON

问题描述

有人对文件接收器连接器的输出格式有疑问吗?

anyone have issues with File Sink Connector output format ?

我想知道为什么我会收到这种格式

I wonder why do I receive this kind of format

{op=u, before={withTax=1, unit=pc(s), deleted=0, updated_at=null, price=420000.0, name=Test 123, created_at=null, id=54, useTaxIncludedFormula=0}, after={withTax=1, unit=pc(s), deleted=0, updated_at=null, price=333.0, name=Test 123, created_at=null, id=54, useTaxIncludedFormula=0}, source={ts_sec=1561038448, query=null, thread=29638, server_id=100596, version=0.9.5.Final, file=mysql-bin.000012, connector=mysql, pos=1044011265, name=crm-kafka-connector, gtid=null, row=0, snapshot=false, db=gxapp_customer_db, table=registrationItemsTmp}, ts_ms=1561063703793}```

但是当我使用 kafka-console-consumer.sh 查看它时，我得到了有效的 JSON

but when I view it using kafka-console-consumer.sh i get valid JSON

{"before":{"id":54,"name":"Test 123","price":420000.0,"unit":"pc(s)","withTax":1,"created_at":null,"updated_at":null,"useTaxIncludedFormula":0,"deleted":0},"after":{"id":54,"name":"Test 123","price":333.0,"unit":"pc(s)","withTax":1,"created_at":null,"updated_at":null,"useTaxIncludedFormula":0,"deleted":0},"source":{"version":"0.9.5.Final","connector":"mysql","name":"crm-kafka-connector","server_id":100596,"ts_sec":1561038448,"gtid":null,"file":"mysql-bin.000012","pos":1044011265,"row":0,"snapshot":false,"thread":29638,"db":"gxapp_customer_db","table":"registrationItemsTmp","query":null},"op":"u","ts_ms":1561063703793}```

我已经设置好一切以使用 JsonConverter

I have setting everything to use JsonConverter

推荐答案

FileStreamSinkConnector 将根据它接收到的值写入数据.JSONConverter 只是告诉 Kafka Connect 如何反序列化它收到的消息.

The FileStreamSinkConnector will write the data based on the values it receives. The JSONConverter just tells Kafka Connect how to deserialise the messages that it receives.

如果您有关于要转储到文件的主题的 JSON，则在接收器连接器中使用

If you have JSON on the topic that you want to just dump to file then in your sink connector use

"value.converter":"org.apache.kafka.connect.storage.StringConverter"

此链接可能会提供您认为有用的更多背景信息:https://rmoff.net/2019/05/08/when-a-kafka-connect-converter-is-not-a-转换器/

This link might provide more background that you find useful: https://rmoff.net/2019/05/08/when-a-kafka-connect-converter-is-not-a-converter/

这篇关于Kafka 文件接收器连接器 - 输出 JSON的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！