3

I'm working on apache flink for data streaming and I have few questions. Any help is greatly appreciated. Thanks.

1) Are there any restrictions on creating tumbling windows. For example, if I want to create a tumbling window per user id for 2 secs and let’s say if I have more than 10 million user id's would that be a problem. (I'm using keyBy user id and then creating a timeWindow for 2 secs)? How are these windows maintained internally in flink?

2) I looked at rebalance for round robin partitioning. Let’s say I have a cluster set up and if I have a parallelism of 1 for source and if I do a rebalance, will my data be shuffled across machines to improve performance? If so is there a specific port using which the data is transferred to other nodes in the cluster?

3) Are there any limitations on state maintenance? I'm planning to maintain some user id related data which could grow very large. I read about flink using rocks db to maintain the state. Just wanted to check if there are any limitations on how much data can be maintained?

4) Also where is the state maintained if the amount of data is less? (I guess in JVM memory) If I have several machines on my cluster can every node get the current state version?

Neoster
  • 195
  • 2
  • 11

1 Answers1

2

  1. If you keyBy your stream on user, Flink will internally partition the stream by users. Hence, the users are distributed across a set of parallel subtasks. The parallelism of the window operator controls the load on each parallel subtask. Handling 10 million users should be no problem if you assign enough machines and configure the parallelism of the program appropriately.

  2. Yes, rebalance() will shuffle over the network if your job runs on multiple machines. With default configuration the data port is automatically chosen. If you need a fixed port, you can use the taskmanager.data.port key to configure it.

  3. The state size limitations depend on the configured state backend. With the RocksDB state backend, the limit is the size of your local filesystem, i.e., RocksDB spills data to disk. In case you hit this limit, you can increase the parallelism because each worker usually handles the key of multiple keys.

  4. It depends on the implementation of the state backend where the state is persisted (disk or memory). I would assume that also the RocksDB state backend which writes to disk caches some data in memory. Please note that operator state is not globally accessible, i.e., each parallel subtask of an operator has only access to its own local state and cannot read or write the state of another subtask of the same operator.

Fabian Hueske
  • 18,707
  • 2
  • 44
  • 49
  • Thanks a lot for the answers. I just have a few followup questions. – Neoster Sep 25 '16 at 17:48
  • If the operator state is not global then lets say if I would like to maintain the previous computation state which is local to a subtask then is there a way to ensure the next data coming in for the same user id goes to the same subtask? If not then do you think instead of maintaining state in flink should I use a centralized cache to achieve this? – Neoster Sep 25 '16 at 18:02
  • Also I'm trying to find a way to send external configuration changes to flink. For example for each computation there are few parameters to be considered. Lets say a new parameter is added and has to be considered for new computations then is there a way to send this change to flink and have like centralized configuration state? – Neoster Sep 25 '16 at 18:03
  • Regarding your first question: If you do `keyBy(user)`, Flink will partition your data and ensure that all records of the same user go to the same subtask. You should prefer key-value state over the `Checkpointed` interface. I will answer your second question on the [new SO question you opened](http://stackoverflow.com/questions/39693919/flink-how-to-handle-external-app-configuration-changes-in-flink). – Fabian Hueske Sep 26 '16 at 07:56
  • If I do a keyBy then apply a window and then do various operations after that, will all the transformations be carried out on the same subtask? Also what will happen if the node goes down and one of the other node picks up this data to process? Just want to better understand the internals. Thanks and appreciate your help. – Neoster Sep 27 '16 at 23:08
  • SO comments are not well suited for longer discussions. I suggest you open new SO questions or write to the Flink user mailing list. – Fabian Hueske Sep 28 '16 at 08:00