shuffle

Documentation

Index

• type Shuffle
  • func NewShuffle(vertexName string, partitionCount int) *Shuffle
  • func (s *Shuffle) ShuffleMessages(messages []*isb.Message) map[int32][]*isb.Message
  • func (s *Shuffle) ShuffleOnId(msgId string) int32
  • func (s *Shuffle) ShuffleOnKeys(keys []string) int32

Types

type Shuffle

type Shuffle struct {
	// contains filtered or unexported fields
}

Shuffle shuffles messages among ISB

func NewShuffle

func NewShuffle(vertexName string, partitionCount int) *Shuffle

NewShuffle accepts list of buffer identifiers(unique identifier of isb) and returns new shuffle instance. It uses vertex-name as seed, without a seed, we will end with the problem where Shuffling before the Vnth vertex creates a key to edge-buffer-index affinity, which will not change from Vn to Vn+1 vertices if there is no re-keying between these vertices causing idle partitions.

func (*Shuffle) ShuffleMessages

func (s *Shuffle) ShuffleMessages(messages []*isb.Message) map[int32][]*isb.Message

ShuffleMessages accepts list of isb messages and returns the mapping of isb to messages

func (*Shuffle) ShuffleOnId

func (s *Shuffle) ShuffleOnId(msgId string) int32

ShuffleOnId shuffle based on the message-id. This is used to make sure we always have consistent hashing for a given message-id. This is used for non-reduce vertex.

func (*Shuffle) ShuffleOnKeys

func (s *Shuffle) ShuffleOnKeys(keys []string) int32

ShuffleOnKeys accepts array of keys and returns a shuffled identifier. We do not need message-id here because there is no multi-partitioning in Reduce streams.