flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Yoandy Rodríguez <mr.domi...@gmail.com>
Subject Kinesis stream and serialization schemas
Date Mon, 02 Sep 2019 14:57:58 GMT
Hi everyone,

As I've mention in previous emails, we're currently exploring flink as a
substitute for some in house products. One of these products sends JSON

data to a Kinesis Data Stream, another product process the records after
some time.

We've tried to set up the Kinesis producer like this:

FlinkKinesisProducer<String>
kinesis=newFlinkKinesisProducer<>(newSimpleStringSchema(), producerConfig);

But on the other application we kept getting some weird binary data, so
right now we're using the following

FlinkKinesisProducer<String>
kinesis=newFlinkKinesisProducer<>(newKinesisSerializationSchema<String>() {
            privatestaticfinallongserialVersionUID=-3435842401751414891L;
            @Override
            publicByteBufferserialize(Stringelement) {
                returnByteBuffer.wrap(element.getBytes());
            }
            @Override
            publicStringgetTargetStream(Stringelement) {
                returnnull;
            }
        }, producerConfig);

But that's not working either. What are we doing wrong?.

Thanks in advance

-- 
Best Regards
Yoandy Rodríguez

Mime
View raw message