Json KAFKA producer

[Luigi Sgaglione <lu...@gmail.com>]

Hi,

I'm trying to create a Flink example with kafka consumer and producer using
Json data format. In particular, I'm able to consume and process Json data
published on a Kafka topic, but not to publish the results.

The problem is that I don't know what is the serialization schema that
should be used to publish an ObjectNode (Jackson).


This is an excerpt of my test code

import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
....

FlinkKafkaConsumer011<ObjectNode> myConsumer = new
FlinkKafkaConsumer011<>("test2", new JSONDeserializationSchema(),
properties);
myConsumer.setStartFromEarliest();

DataStreamSource<ObjectNode> stream = env.addSource(myConsumer);
SingleOutputStreamOperator<ObjectNode> out1 = stream.filter(new
FilterFunction<ObjectNode>() {
private static final long serialVersionUID = 1L;

@Override
public boolean filter(ObjectNode arg0) throws Exception {
String temp=arg0.get("value").asText();
return (!temp.equals("1"));
}
});
FlinkKafkaProducer011<ObjectNode> producer = new
FlinkKafkaProducer011<ObjectNode>("192.168.112.128:9092", "flinkOut", *XXX*
);
out1.addsink(producer);

Can you help me to understand how I can publish an ObjectNode?

Thanks

Re: Json KAFKA producer

[Fabian Hueske <fh...@gmail.com>]

Hi,

SerializationSchema is a public interface that you can implement.
It has a single method to turn an object into a byte array.

I would suggest to implement your own SerializationSchema.

Best, Fabian

2018-04-11 15:56 GMT+02:00 Luigi Sgaglione <lu...@gmail.com>:

> Hi,
>
> I'm trying to create a Flink example with kafka consumer and producer
> using Json data format. In particular, I'm able to consume and process Json
> data published on a Kafka topic, but not to publish the results.
>
> The problem is that I don't know what is the serialization schema that
> should be used to publish an ObjectNode (Jackson).
>
>
> This is an excerpt of my test code
>
> import org.apache.flink.shaded.jackson2.com.fasterxml.
> jackson.databind.node.ObjectNode;
> ....
>
> FlinkKafkaConsumer011<ObjectNode> myConsumer = new
> FlinkKafkaConsumer011<>("test2", new JSONDeserializationSchema(),
> properties);
> myConsumer.setStartFromEarliest();
>
> DataStreamSource<ObjectNode> stream = env.addSource(myConsumer);
> SingleOutputStreamOperator<ObjectNode> out1 = stream.filter(new
> FilterFunction<ObjectNode>() {
> private static final long serialVersionUID = 1L;
>
> @Override
> public boolean filter(ObjectNode arg0) throws Exception {
> String temp=arg0.get("value").asText();
> return (!temp.equals("1"));
> }
> });
> FlinkKafkaProducer011<ObjectNode> producer = new FlinkKafkaProducer011<
> ObjectNode>("192.168.112.128:9092", "flinkOut", *XXX*);
> out1.addsink(producer);
>
> Can you help me to understand how I can publish an ObjectNode?
>
> Thanks
>
>
>