Hi,
void output(DataStream<List<SuperclassEvent>> inputStream) {
These seems odd. Are your events intended to be a list? If not, this should be
a `DataStream<SuperclassEvent>`.
From the code snippet you’ve attached in the first post, it seems like you’ve
initialized your source incorrectly.
`env.fromElements(List<...>)` will take the whole list as a single event, thus
your source is only emitting a single list as a record. Perhaps what you
actually want to do here is `env.fromCollection(List<...>)`?
This should also eliminate the situation that “only after the whole List is
processed can the records then be sent to their respective sinks”, as you
mentioned in your reply.
but this doesn't seem very ideal to us because it requires a new operator to
first split the stream
IMO, this wouldn’t really introduce noticeable overhead, as the operator will
be chained to the map operator. Side outputs is also the preferred way here, as
side outputs subsume stream splitting.
Overall, I think it is reasonable to do a map -> split -> Kafka / Cassandra
sinks in your case, given that you’ve declared the source correctly to be a
single SuperclassEvent as a record.
The operator overhead is fairly trivial if it is chained. Another reason to use
sinks properly is that only then will you benefit from the exactly-once /
at-least-once delivery guarantees to external systems (which requires
collaboration between the sink and Flink’s checkpointing).
Hope this helps!
Cheers,
Gordon
On 17 July 2017 at 2:59:38 AM, earellano [via Apache Flink User Mailing List
archive.] (ml+s2336050n14294...@n4.nabble.com) wrote:
Tzu-Li (Gordon) Tai wrote
It seems like you’ve misunderstood how to use the FlinkKafkaProducer, or is
there any specific reason why you want to emit elements to Kafka in a map
function?
The correct way to use it is to add it as a sink function to your pipeline, i.e.
DataStream<String> someStream = …
someStream.addSink(new FlinkKafkaProducer010<>(“topic”, schema, props));
// or, FlinkKafkaProducer010.writeToKafkaWithTimestamps(someStream, “topic”,
schema, props);
The reason we want to use processElement() & a map function, instead of
someStream.addSink() is that our output logic has conditional depending on the
type of record we have.
Our overall program follows this path:
Serialized JSON consumed from Kafka: DataStream<byte []>
parsed, producing a List of successful events and/or error events:
DataStream<List<Events>>
outputted conditionally, going to Kafka or Cassandra depending on which type
of event it is.
This is our code for output logic (although modified types to not use our IP):
void output(DataStream<List<SuperclassEvent>> inputStream) {
inputStream.map( eventList ->
for (SuperclassEvent event : eventList) {
if (event instanceof SuccessEvent)
emitToCassandra(event);
else if (event instanceof ErrorEvent)
emitToKafka(event);
}
return true; // we don't actually want to return anything, just don't
know how else to use map
);
That is, we have sinks for both Kafka and Cassandra, and want to be able to
iterate through our List<SuperclassElement> and conditionally send each
individual record to its appropriate sink depending on its type.
I know Flink offers SplitStreams for a similar purpose, but this doesn't seem
very ideal to us because it requires a new operator to first split the stream,
and only after the whole List is processed can the records then be sent to
their respective sinks. Whereas the code above sends the records to their sinks
immediately upon finding its type.
--
Is there any way to make processElement() work so that we can work on
individual records instead of the whole DataStream? Or are we misusing Flink?
How do you recommend doing this the best way possible?
--
Also, if processElement() and invoke() aren't meant to be used, should they be
made package private? Happy to make a pull request if so, although fear that
might break a few things.
If you reply to this email, your message will be added to the discussion below:
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Kafka-Producer-Null-Pointer-Exception-when-processing-by-element-tp14288p14294.html
This email was sent by earellano (via Nabble)
To receive all replies by email, subscribe to this discussion
