从Kafka到GCS Bucket的Apache波束流数据(不使用pubsub)

Question

我见过很多Apache Beam的例子，从PubSub读取数据并写入GCS bucket，但是有使用KafkaIO并将其写入GCS bucket的例子吗？我可以在哪里解析消息并根据消息内容将其放入适当的桶中？

对于例如。

message = {type="type_x", some other attributes....}
message = {type="type_y", some other attributes....}

type_x --> goes to bucket x
type_y --> goes to bucket y

我的用例是将数据从 Kafka 流式传输到 GCS 存储桶，所以如果有人建议在 GCP 中提供更好的方法，它也欢迎。

谢谢。问候，阿南特。

Answer 1

你可以看看这里的例子-https://github.com/0x0ece/beam-starter/blob/master/src/main/java/com/dataradiant/beam/examples/StreamWordCount.java

一旦您读取了数据元素，如果您想基于特定数据值写入多个目标，则可以使用TupleTagList查看多个输出，详细信息可在此处找到-https://beam.apache.org/documentation/programming-guide/#additional-输出

Answer 2

您可以使用 Secor 将消息加载到 GCS 存储桶。Secor还能够解析传入的消息，并将它们放在同一个存储桶中的不同路径下。