我有一个kafka0.10的运行示例，目前正在使用gobblin将数据存储到hdfs中。我想切换到Kafka连接，我研究发现，合流提供了一个连接器。但是，有没有一种方法可以在不使用整个汇合平台的情况下使用此连接器？例如，我可以从合流源复制相关脚本，并以某种方式使我的Kafka示例使用它吗？我基本上还在通过这些东西学习我的方法，所以我还不是很精通这个领域。谢谢。
8aqjt8rx1#
是的，这是可能的。我已经做到了。我使用在docker容器中运行的稍微修改过的confluent hdfs独立连接器。但是，您也必须使用schemaregistry。因为连接器与schemaregistry紧密耦合。此外，您还必须发送特殊格式的邮件。为了支持自动模式识别，合流kafka消费者引入了一种内部消息格式。因此，为了与合流消费者兼容，生产者必须按照以下格式编写消息。标题（5字节）消息“magic byte”的第一个字节应始终为0接下来的4个字节应该是以big-endian格式编码的schema registry中schema的id。有效载荷（avro\parquet对象，二进制编码）。ps向topic发送消息时要非常小心，因为如果消息与schema不匹配，或者注册表中不存在id为的schema，则使用者会无声地失败：工作线程停止，但应用程序仍挂起在内存中，不会退出。
1条答案
按热度按时间8aqjt8rx1#
是的，这是可能的。我已经做到了。我使用在docker容器中运行的稍微修改过的confluent hdfs独立连接器。但是，您也必须使用schemaregistry。因为连接器与schemaregistry紧密耦合。此外，您还必须发送特殊格式的邮件。为了支持自动模式识别，合流kafka消费者引入了一种内部消息格式。因此，为了与合流消费者兼容，生产者必须按照以下格式编写消息。
标题（5字节）
消息“magic byte”的第一个字节应始终为0
接下来的4个字节应该是以big-endian格式编码的schema registry中schema的id。
有效载荷（avro\parquet对象，二进制编码）。
ps向topic发送消息时要非常小心，因为如果消息与schema不匹配，或者注册表中不存在id为的schema，则使用者会无声地失败：工作线程停止，但应用程序仍挂起在内存中，不会退出。