No accumulator results in streaming
Locked
No accumulator results in streaming
 
|
Hello, I’m struggling with this simple issue for hours now : I am unable to get the accumulator result of a streaming context result, the accumulator map
in the JobExecutionResult is always empty. Simple test code (directly inspired from the documentation) : My source =
public
static
class
oneRandomNumberSource implements SourceFunction<Integer>, Serializable {
@Override
public
void run(org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<Integer>
ctx)
throws Exception {
final Random
rnd =
new Random(29172);
ctx.collect(rnd.nextInt()); }
@Override
public
void cancel() {
} } My exec program =
public
static
final String
COUNTER_NBLINE =
"num-lines";
void test() { final
StreamExecutionEnvironment env = getCluster();
final SourceFunction<Integer>
source =
new oneRandomNumberSource();
env.addSource(source).addSink(new
RichSinkFunction<Integer>() {
private IntCounter
numLines =
new IntCounter();
@Override
public
void open(Configuration
parameters)
throws Exception {
// NOPMD getRuntimeContext().addAccumulator(COUNTER_NBLINE,
this.numLines); }
@Override
public
void invoke(Integer
value)
throws Exception { System.err.println(value);
numLines.add(1); } });
try {
final JobExecutionResult
result =
env.execute(); System.out.println(result.getAccumulatorResult(COUNTER_NBLINE));
// Problem : always null }
catch (Exception
e) {
e.printStackTrace();
} } Console output : 07/16/2015 14:11:58 Job execution switched to status RUNNING. 07/16/2015 14:11:58 Custom Source(1/1) switched to SCHEDULED
07/16/2015 14:11:58 Custom Source(1/1) switched to DEPLOYING
07/16/2015 14:11:58 Stream Sink(1/4) switched to SCHEDULED
07/16/2015 14:11:58 Stream Sink(1/4) switched to DEPLOYING
07/16/2015 14:11:58 Stream Sink(2/4) switched to SCHEDULED
07/16/2015 14:11:58 Stream Sink(2/4) switched to DEPLOYING
07/16/2015 14:11:58 Stream Sink(3/4) switched to SCHEDULED
07/16/2015 14:11:58 Stream Sink(3/4) switched to DEPLOYING
07/16/2015 14:11:58 Stream Sink(4/4) switched to SCHEDULED
07/16/2015 14:11:58 Stream Sink(4/4) switched to DEPLOYING
07/16/2015 14:11:58 Custom Source(1/1) switched to RUNNING
07/16/2015 14:11:58 Stream Sink(1/4) switched to RUNNING
07/16/2015 14:11:58 Stream Sink(2/4) switched to RUNNING
07/16/2015 14:11:58 Stream Sink(4/4) switched to RUNNING
07/16/2015 14:11:58 Stream Sink(3/4) switched to RUNNING
07/16/2015 14:11:58 Custom Source(1/1) switched to FINISHED
07/16/2015 14:11:58 Stream Sink(4/4) switched to FINISHED
07/16/2015 14:11:58 Stream Sink(3/4) switched to FINISHED
-329782788 07/16/2015 14:11:58 Stream Sink(2/4) switched to FINISHED
07/16/2015 14:11:58 Stream Sink(1/4) switched to FINISHED
07/16/2015 14:11:58 Job execution switched to status FINISHED. null What am I doing wrong ? Flink version is 0.9.0 Best regards, Arnaud L'intégrité de ce message n'étant pas assurée sur internet, la société expéditrice ne peut être tenue responsable de son contenu ni de ses pièces jointes. Toute utilisation ou diffusion non autorisée est interdite. Si vous n'êtes pas destinataire de ce message, merci de le détruire et d'avertir l'expéditeur. The integrity of this message cannot be guaranteed on the Internet. The company that sent this message cannot therefore be held liable for its content nor attachments. Any unauthorized use or dissemination is prohibited. If you are not the intended recipient of this message, then please delete it and notify the sender. |
Re: No accumulator results in streaming
|
|
Hi Arnaud! In 0.9.0, the streaming API does not forward accumulators. In the next version, it will, and it will actually update them live so that you can retrieve continuously updating accumulators of a streaming job in your client, and in the web frontend. We merged the first part for that on Wednesday, so there should be support for that in the snapshot next week. Greetings, Stephan On Thu, Jul 16, 2015 at 2:21 PM, LINZ, Arnaud <[hidden email]> wrote:
|
| Free forum by Nabble | Edit this page |