Springxd从rabbitmq json消息转换为自定义火花处理器模块中的pojo

Question

我正在尝试创建如下流：

rabbitmq |火花定制处理器|文件接收器

RabbitMq 从包含消息的队列中读取，其标头类型为 application/json，代表我的 POJO 类 House。然后这些读取的消息应该被传输到我的spark-custom-processor 并且在处理器完成他的工作之后，将结果存储在一个文件中。

问题是到达spark-custom-processor的数据类型是Byte而不是转换为我的POJO类House的application/json消息。

如果阅读解释 Type convertion 的文档，并且它表明转换为我的 POJO House 类型应该自动完成，但它没有。

我也试过用：

rabbitmq | --inputputType=application/x-java-object;type=domain.MetricBean spark-custom-processor |文件接收器 指定转换，但 springXdShell 抱怨 ; 处的 sintax 错误

有没有办法做到这一点，或者我必须自己在我的spark-custom-processor 中使用ObjectMapper 进行映射？

Answer 1

你只需要用;引用字符串。

我刚刚测试了这个，效果很好......

xd:>stream create foo --definition "rabbit --outputType='application/x-java-object;type=foo.Foo' 
    | log" --deploy

.

public class Foo {

    private String foo;

    public String getFoo() {
        return foo;
    }

    public void setFoo(String foo) {
        this.foo = foo;
    }

    @Override
    public String toString() {
        return "Foo [foo=" + foo + "]";
    }

}

已发布{"foo":"bar"}

2015-07-30T10:37:36-0400 1.2.0.RC1 INFO SimpleAsyncTaskExecutor-1 sink.foo - Foo [foo=bar]