Edit - Stack Overflow

You are not logged in. Your edit will be placed in a queue until it is peer reviewed.

We welcome edits that make the post easier to understand and more valuable for readers. Because community members review edits, please try to make the post substantially better than how you found it, for example, by fixing grammar or adding additional resources and hyperlinks.

Required fields*

Rev

Required fields*

Buffering transformed messages(example, 1000 count) using Apache Flink stream processing

I'm using Apache Flink for stream processing.

After subscribing the messages from source(ex:Kafka, AWS Kinesis Data Streams) and then applying transformation, aggregation and etc. using Flink operators on streaming data I want to buffer final messages(ex:1000 in count) and post each batch in a single request to external REST API.

How to implement buffering mechanism(creating each 1000 records as a batch) in Apache Flink?

Flink pipileine: streaming Source --> transform/reduce using Operators --> buffer 1000 messages --> post to REST API

Appreciate your help!

Answer*

I'd create a sink with state that would hold on to the messages that are passed in. When the count gets high enough (1000) the sink sends the batch. The state can be in memory (e.g. an instance variable holding an ArrayList of messages), but you should use checkpoints so that you can recover that state in case of a failure of some kind.

When your sink has checkpointed state, it needs to implement CheckpointedFunction (in org.apache.flink.streaming.api.checkpoint) which means you need to add two methods to your sink:

	@Override
	public void snapshotState(FunctionSnapshotContext context) throws Exception {
		
		checkpointedState.clear();

		// HttpSinkStateItem is a user-written class 
		// that just holds a collection of messages (Strings, in this case)
		//
		// Buffer is declared as ArrayList<String>
		
		checkpointedState.add(new HttpSinkStateItem(buffer));

	}

	@Override
	public void initializeState(FunctionInitializationContext context) throws Exception {

		// Mix and match different kinds of states as needed:
		// - Use context.getOperatorStateStore() to get basic (non-keyed) operator state
		// - types are list and union		
		// - Use context.getKeyedStateStore() to get state for the current key (only for processing keyed streams)
		// - types are value, list, reducing, aggregating and map
		// - Distinguish between state data using state name (e.g. "HttpSink-State")		
		
		ListStateDescriptor<HttpSinkStateItem> descriptor =
 new ListStateDescriptor<>(
 "HttpSink-State",
 HttpSinkStateItem.class);

 checkpointedState = context.getOperatorStateStore().getListState(descriptor);

 if (context.isRestored()) {

 	for (HttpSinkStateItem item: checkpointedState.get()) {
 			buffer = new ArrayList<>(item.getPending()); 
 	}

 	}		

	}
 
 

You can also use a timer in the sink (if the input stream is keyed/partitioned) to send periodically if the count doesn't reach your threshold.

Draft saved

Draft discarded

Edit Summary*

Cancel

Chris, are you referring "Keyed State" as state, for creating sinc with and hold on to messages? If you have a reference example please link, I’m new to Flink, but anyway I will read docs. Thank you!

Sparkle8
– Sparkle8

2019年11月07日 14:57:19 +00:00
Commented Nov 7, 2019 at 14:57
@Sparkle8 I'll see what I can come up with

Chris Gerken
– Chris Gerken

2019年11月08日 08:33:17 +00:00
Commented Nov 8, 2019 at 8:33
thank you, Chris! your answer is helpful for implementing session based stream processing and post on messages to REST, with apache-flink

Sparkle8
– Sparkle8

2019年11月11日 17:08:34 +00:00
Commented Nov 11, 2019 at 17:08

Add a comment |

How to Edit

Correct minor typos or mistakes
Clarify meaning without changing it
Add related resources or links
Always respect the author’s intent
Don’t use edits to reply to the author

How to Format

create code fences with backticks ` or tildes ~
```
like so
```
add language identifier to highlight code
```python
def function(foo):
print(foo)
```
put returns between paragraphs
for linebreak add 2 spaces at end
_italic_ or **bold**
indent code by 4 spaces
backtick escapes `like _so_`
quote by placing > at start of line
to make links (use https whenever possible)

<https://example.com>

[example](https://example.com)

<a href="https://example.com">example</a>

formatting help »
answering help »

How to Tag

A tag is a keyword or label that categorizes your question with other, similar questions. Choose one or more (up to 5) tags that will help answerers to find and interpret your question.

complete the sentence: my question is about...
use tags that describe things or concepts that are essential, not incidental to your question
favor using existing popular tags
read the descriptions that appear below the tag

If your question is primarily about a topic for which you can't find a tag:

combine multiple words into single-words with hyphens (e.g. python-3.x), up to a maximum of 35 characters
creating new tags is a privilege; if you can't yet create a tag you need, then post this question without it, then ask the community to create it for you

popular tags »

CollectivesTM on Stack Overflow

Buffering transformed messages(example, 1000 count) using Apache Flink stream processing

Answer*