5

我写了一个风暴拓扑。我基本上想以字节数组的形式将 avro 模式中的元组发送到 kafka 主题。

这就是我设置螺栓的方式：

  builder.setBolt(KAFKA_AVRO_BOLT_NAME, new KafkaBolt<String, byte[]>())
            .fieldsGrouping(BOLT1, new Fields("key"));

这就是我转换为字节数组的方式

Schema schema = avroObject.getSchema();

        DatumWriter<GenericRecord> writer = new GenericDatumWriter<GenericRecord>(schema);
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
        writer.write(ping, encoder);
        encoder.flush();
        byte[] message = out.toByteArray();
        String key = new String(message, "UTF-8");

当我以以下方式发出元组时，我在 kafka 主题中看不到任何内容（向 kafka 发送字节流）：

collector.emit(tuple, new Values(Obj.hashMD5(key), message));

但相反，如果我将字节数组转换为字符串，然后转换为 kafka 主题，它可以工作：

如下所示：

 builder.setBolt(KAFKA_AVRO_BOLT_NAME, new KafkaBolt<String, String>())
            .fieldsGrouping(BOLT1, new Fields("key"));

collector.emit(tuple, new Values(Obj.hashMD5(key), key));

我究竟做错了什么？如何使用 Storm kafka bolt 将字节流发送到 kafka 主题？

4

1 回答 1

5

您遇到问题是因为您的 MD5 哈希不正确：

您说，如果您将 bytearray 转换为 java String 它可以工作：这是因为 MD5 的值根据 String 是正确的。

collector.emit(tuple, new Values(Obj.hashMD5(key), key));

如您所见，MD5 是根据字符串参数计算的，您发送与 MD5 对应的字符串：一切都很好！

但是如果你发送一个字节数组，你需要在一个字节数组上计算 MD5，结果它将是一个字节数组，而不是一个字符串。你的代码：

collector.emit(tuple, new Values(Obj.hashMD5(key), message));

不正确，因为 MD5 不对应于消息，而是对应于 UTF-8 中消息的转换值作为有损字符串（见下文）。

这是一个关于 SO 的另一个问题的链接，以字节数组格式正确计算 MD5：

如何生成 MD5 哈希？

这是因为在 Java 中将 bytearray 转换为 String 是有损的（与 C 相反），并且您将在该过程中丢失数据，因为某些字节与 Java 编码中的 char 不对应（您的数据中显然有其中一些）。

所以你的 KafkaBolt 应该是

KafkaBolt<byte[], byte[]>

我不知道在 kafka Storm 中发送一个 bytearray MD5 和你的 bytearray 是否足够。如果不是，则必须在 bytearray 和 java String 之间使用无损编码，例如 BASE64：

Java中的Base64编码

您必须使用将 bytearray 转换为 base64 字符串

KafkaBolt<String, String>

然后像往常一样发送数据

collector.emit(tuple, new Values(Obj.hashMD5(keyInBase64), keyInBase64));

这也意味着当您从 kafka 获取数据时，它将是 base64 中的字符串，您必须对其进行解码才能取回字节数组。

希望有帮助。

于 2015-03-18T09:14:57.133 回答