Method Map

flow/dataset_map.go:19–33

map can work with multiple kinds of inputs and outputs Input Types: 1. single value 2. (key, value) : Most common format for key value pair 3. (key, values) : GroupByKey() outputs 4. (key, values1, values2) : CoGroup() outputs 5. (key, value1, value2) : Join() outputs Output Types: 1. return single

(f interface{})

//   3. return no value
//   4. return no value, but last parameter is a output channel
func (d *Dataset) Map(f interface{}) *Dataset {
	outType := guessFunctionOutputType(f)
	ret, step := add1ShardTo1Step(d, outType)
	step.Name = "Map"
	step.Function = func(task *Task) {

		invokeMapFunc := _buildMapperFunction(f, task)

		for input := range task.InputChan() {
			invokeMapFunc(input)
		}
		// println("exiting d:", d.Id, "step:", step.Id, "task:", task.Id)
	}
	return ret
}

func _buildMapperFunction(f interface{}, task *Task) func(input reflect.Value) {
	fn, ft := reflect.ValueOf(f), reflect.TypeOf(f)