1. 程式人生 > >聊聊storm的CustomStreamGrouping

聊聊storm的CustomStreamGrouping

本文主要研究一下storm的CustomStreamGrouping

CustomStreamGrouping

storm-2.0.0/storm-client/src/jvm/org/apache/storm/grouping/CustomStreamGrouping.java

public interface CustomStreamGrouping extends Serializable {

    /**
     * Tells the stream grouping at runtime the tasks in the target bolt. This information should be used in
chooseTasks to determine the * target tasks. * * It also tells the grouping the metadata on the stream this grouping will be used on. */ void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks); /** * This function implements a custom stream grouping. It takes in
as input the number of tasks in the target bolt in prepare and returns * the tasks to send the tuples to. * * @param values the values to group on */ List<Integer> chooseTasks(int taskId, List<Object> values); } 複製程式碼
  • 這裡定義了prepare以及chooseTasks方法
  • GrouperFactory裡頭定義了FieldsGrouper、GlobalGrouper、NoneGrouper、AllGrouper、BasicLoadAwareCustomStreamGrouping
  • 另外org.apache.storm.grouping包裡頭也定義了ShuffleGrouping、PartialKeyGrouping、LoadAwareShuffleGrouping

FieldsGrouper

storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/GrouperFactory.java

    public static class FieldsGrouper implements CustomStreamGrouping {

        private Fields outFields;
        private List<List<Integer>> targetTasks;
        private Fields groupFields;
        private int numTasks;

        public FieldsGrouper(Fields outFields, Grouping thriftGrouping) {
            this.outFields = outFields;
            this.groupFields = new Fields(Thrift.fieldGrouping(thriftGrouping));

        }

        @Override
        public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks) {
            this.targetTasks = new ArrayList<List<Integer>>();
            for (Integer targetTask : targetTasks) {
                this.targetTasks.add(Collections.singletonList(targetTask));
            }
            this.numTasks = targetTasks.size();
        }

        @Override
        public List<Integer> chooseTasks(int taskId, List<Object> values) {
            int targetTaskIndex = TupleUtils.chooseTaskIndex(outFields.select(groupFields, values), numTasks);
            return targetTasks.get(targetTaskIndex);
        }

    }
複製程式碼
  • 對選中fields的values通過TupleUtils.chooseTaskIndex選擇task下標;chooseTaskIndex主要是採用Arrays.deepHashCode取雜湊值然後對numTask向下取模

GlobalGrouper

storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/GrouperFactory.java

    public static class GlobalGrouper implements CustomStreamGrouping {

        private List<Integer> targetTasks;

        public GlobalGrouper() {
        }

        @Override
        public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks) {
            this.targetTasks = targetTasks;
        }

        @Override
        public List<Integer> chooseTasks(int taskId, List<Object> values) {
            if (targetTasks.isEmpty()) {
                return null;
            }
            // It's possible for target to have multiple tasks if it reads multiple sources
            return Collections.singletonList(targetTasks.get(0));
        }
    }
複製程式碼
  • 這裡固定取第一個task,即targetTasks.get(0)

NoneGrouper

storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/GrouperFactory.java

    public static class NoneGrouper implements CustomStreamGrouping {

        private final Random random;
        private List<Integer> targetTasks;
        private int numTasks;

        public NoneGrouper() {
            random = new Random();
        }

        @Override
        public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks) {
            this.targetTasks = targetTasks;
            this.numTasks = targetTasks.size();
        }

        @Override
        public List<Integer> chooseTasks(int taskId, List<Object> values) {
            int index = random.nextInt(numTasks);
            return Collections.singletonList(targetTasks.get(index));
        }
    }
複製程式碼
  • 這裡通過random.nextInt(numTasks)隨機取task

AllGrouper

storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/GrouperFactory.java

    public static class AllGrouper implements CustomStreamGrouping {

        private List<Integer> targetTasks;

        @Override
        public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks) {
            this.targetTasks = targetTasks;
        }

        @Override
        public List<Integer> chooseTasks(int taskId, List<Object> values) {
            return targetTasks;
        }
    }
複製程式碼
  • 這裡返回所有的targetTasks

ShuffleGrouping

storm-2.0.0/storm-client/src/jvm/org/apache/storm/grouping/ShuffleGrouping.java

public class ShuffleGrouping implements CustomStreamGrouping, Serializable {
    private ArrayList<List<Integer>> choices;
    private AtomicInteger current;

    @Override
    public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks) {
        choices = new ArrayList<List<Integer>>(targetTasks.size());
        for (Integer i : targetTasks) {
            choices.add(Arrays.asList(i));
        }
        current = new AtomicInteger(0);
        Collections.shuffle(choices, new Random());
    }

    @Override
    public List<Integer> chooseTasks(int taskId, List<Object> values) {
        int rightNow;
        int size = choices.size();
        while (true) {
            rightNow = current.incrementAndGet();
            if (rightNow < size) {
                return choices.get(rightNow);
            } else if (rightNow == size) {
                current.set(0);
                return choices.get(0);
            }
        } // race condition with another thread, and we lost. try again
    }
}
複製程式碼
  • 這裡在prepare的時候對ArrayList<List> choices進行隨機化
  • 採用current.incrementAndGet()實現round robbin的效果,超過size的時候重置返回第一個,沒有超過則返回incr後的index的值

PartialKeyGrouping

storm-2.0.0/storm-client/src/jvm/org/apache/storm/grouping/PartialKeyGrouping.java

public class PartialKeyGrouping implements CustomStreamGrouping, Serializable {
    private static final long serialVersionUID = -1672360572274911808L;
    private List<Integer> targetTasks;
    private Fields fields = null;
    private Fields outFields = null;

    private AssignmentCreator assignmentCreator;
    private TargetSelector targetSelector;

    public PartialKeyGrouping() {
        this(null);
    }

    public PartialKeyGrouping(Fields fields) {
        this(fields, new RandomTwoTaskAssignmentCreator(), new BalancedTargetSelector());
    }

    public PartialKeyGrouping(Fields fields, AssignmentCreator assignmentCreator) {
        this(fields, assignmentCreator, new BalancedTargetSelector());
    }

    public PartialKeyGrouping(Fields fields, AssignmentCreator assignmentCreator, TargetSelector targetSelector) {
        this.fields = fields;
        this.assignmentCreator = assignmentCreator;
        this.targetSelector = targetSelector;
    }

    @Override
    public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks) {
        this.targetTasks = targetTasks;
        if (this.fields != null) {
            this.outFields = context.getComponentOutputFields(stream);
        }
    }

    @Override
    public List<Integer> chooseTasks(int taskId, List<Object> values) {
        List<Integer> boltIds = new ArrayList<>(1);
        if (values.size() > 0) {
            final byte[] rawKeyBytes = getKeyBytes(values);

            final int[] taskAssignmentForKey = assignmentCreator.createAssignment(this.targetTasks, rawKeyBytes);
            final int selectedTask = targetSelector.chooseTask(taskAssignmentForKey);

            boltIds.add(selectedTask);
        }
        return boltIds;
    }
    //......
}
複製程式碼
  • 這裡通過RandomTwoTaskAssignmentCreator來選中兩個taskId,然後選擇使用次數小的那個

LoadAwareCustomStreamGrouping

storm-2.0.0/storm-client/src/jvm/org/apache/storm/grouping/LoadAwareCustomStreamGrouping.java

public interface LoadAwareCustomStreamGrouping extends CustomStreamGrouping {
    void refreshLoad(LoadMapping loadMapping);
}
複製程式碼
  • 繼承了CustomStreamGrouping介面,然後新定義了refreshLoad方法用於重新整理負載,這裡的負載主要是executor的receiveQueue的負載(qMetrics.population() / qMetrics.capacity())
  • LoadAwareCustomStreamGrouping有幾個實現類,有BasicLoadAwareCustomStreamGrouping以及LoadAwareShuffleGrouping

小結

  • storm的CustomStreamGrouping介面定義了chooseTasks方法,用於選擇tasks來處理tuples
  • ShuffleGrouping類似round robbin,FieldsGrouper則根據所選欄位值採用Arrays.deepHashCode取雜湊值然後對numTask向下取模,GlobalGrouper返回index為0的taskId,NoneGrouper則隨機返回,AllGrouper不做過濾返回所有taskId,PartialKeyGrouping則使用key的雜湊值作為seed,採用Random函式來計算兩個taskId的下標,然後選擇使用次數少的那個task。
  • LoadAware的grouping有BasicLoadAwareCustomStreamGrouping以及LoadAwareShuffleGrouping,他們都實現了LoadAwareCustomStreamGrouping介面,該介面定義了refreshLoad方法,用於動態重新整理負載,這裡的負載主要是executor的receiveQueue的負載(qMetrics.population() / qMetrics.capacity())

doc