[jira] [Comment Edited] (EAGLE-24) Load skew issue when partition by user

["Edward Zhang (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/EAGLE-24?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15004630#comment-15004630 ] 

Edward Zhang edited comment on EAGLE-24 at 11/13/15 8:11 PM:
-------------------------------------------------------------

This problem is actually data skew problem compared to compute skew problem which is elaborated on https://issues.apache.org/jira/browse/EAGLE-21

The problem can be simplified as put objects with weight in different bins and make sure each bin has approximately even total weight.

Let U={u1,u2,...,un} be a set of LONG weights which is messages per second for this user. Let B={b1,b2,...bm} be a set of following bolts, with m≤n. Let T(bj) represent the total weight present in bolt bj, which is the sum of all the weights present in bj.

What is the optimal way to distribute all the weights wi into the buckets bj to minimize the metric maxT(bj)−minT(bk) for some j,k≤n?

Use greedy algorithm for this partition problem. https://en.wikipedia.org/wiki/Partition_problem#The_greedy_algorithm

But in practical we should consider the following:
1. User's weight should be exposed as a service, so in runtime, each node can download user's weight and determine what is the following node this user should go
2. How to update user's weight. Each node should do statistics and put the result into service.
3. Each node should use the same user-weight table
4. Each node should generate the same route table in memory based on the same user-weight table
5. When node refreshes user-weight table and uses the new route table, then it would impact the following nodes if the following node is stateful in processing one user, for example it has a slide window rule which is grouped by user
6. when new user comes, it should obey the rule of evenly hashing


was (Author: yonzhang2012):
This problem is actually data skew problem compared to compute skew problem which is elaborated on https://issues.apache.org/jira/browse/EAGLE-21

The problem can be simplified as put objects with weight in different bins and make sure each bin has approximately even total weight.

Let U={u1,u2,...,un} be a set of LONG weights which is messages per second for this user. Let B={b1,b2,...bm} be a set of following bolts, with m≤n. Let T(bj) represent the total weight present in bolt bj, which is the sum of all the weights present in bj.

What is the optimal way to distribute all the weights wi into the buckets bj to minimize the metric maxT(bj)−minT(bk) for some j,k≤n?

Use greedy algorithm for this partition problem. https://en.wikipedia.org/wiki/Partition_problem#The_greedy_algorithm

But in practical we should consider the following:
1. User's weight should be exposed as a service, so in runtime, each node can download user's weight and determine what is the following node this user should go
2. How to update user's weight. Each node should do statistics and put the result into service.
3. Each node should use the same user-weight table
4. Each node should generate the same route table in memory based on the same user-weight table
5. When node refreshes user-weight table and uses the new route table, then it would impact the following nodes if the following node is stateful in processing one user, for example it has a slide window rule which is grouped by user

> Load skew issue when partition by user
> --------------------------------------
>
>                 Key: EAGLE-24
>                 URL: https://issues.apache.org/jira/browse/EAGLE-24
>             Project: Eagle
>          Issue Type: Improvement
>            Reporter: Libin, Sun
>            Assignee: Libin, Sun
>
> Today hdfs audit log/hive query log events are partitioned by user field and pass all the way from source to spouts, bolts
> But due to the existence of heavy user, the load of spout/bolt are skewed, especially for alert executors which may be computing/memory intensive
> We need to resolve this by dynamic partition events based on spout/bolt capacity



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)