import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(String key, Person value, Context<String, Integer> context) {
    if (state != null) {
        if (!state.equals(value.getState())) {
            return;
        }
    }

    String name = abbrevationMapping.get(value.getState());

    IMap<String, SalaryYear> salariesMap = hazelcastInstance.getMap("salaries");

    // Person's partition key is email!
    SalaryYear salaryYear = salariesMap.get(value.getEmail());

    for (SalaryMonth salaryMonth : salaryYear.getMonths()) {
        context.emit(name, salaryMonth.getSalary());
    }
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(
        final Integer key,
        final SpatialPoint value,
        final Context<Integer, Integer> collector)
{
    final double x = value.point.getX();
    final double y = value.point.getY();

    m_insidePolygon.setXY(x, y);

    final Collection<SpatialPolygon> collection = m_map.values(m_insidePolygon);
    if (!collection.isEmpty())
    {
        collector.emit(collection.iterator().next().id, 1);
    }
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(String key, String value, Context<String, Integer> context) {
    StringTokenizer tokenizer = new StringTokenizer(value);
    while (tokenizer.hasMoreTokens()) {
        String word = WordCountExample.cleanWord(tokenizer.nextToken());
        if (word.length() >= 4) {
            context.emit(word.toLowerCase(), ONE);
        }
    }
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void initialize(Context<String, Integer> context) {
    super.initialize(context);

    IMap<Integer, State> states = hazelcastInstance.getMap("states");
    for (State s : states.values()) {
        String abbrevation = s.getAbbreviation();
        String name = s.getName();
        abbrevationMapping.put(abbrevation, name);
    }
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(String key, Person value, Context<String, Person> context) {
    if (state != null) {
        if (!state.equals(value.getState())) {
            return;
        }
    }

    context.emit(value.getState(), value);
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(String key, Crime value, Context<CrimeCategory, Integer> context) {
    if (!stateName.equals(value.getState())) {
        return;
    }

    context.emit(value.getCrime(), value.getCount());
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(String key, Person value, Context<String, Integer> context) {
    if (state != null) {
        if (!state.equals(value.getState())) {
            return;
        }
    }

    String name = abbrevationMapping.get(value.getState());
    context.emit(name, ONE);
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(String key, String value, Context<String, Integer> context) {
    StringTokenizer tokenizer = new StringTokenizer(value);
    while (tokenizer.hasMoreTokens()) {
        String word = WordCount.cleanWord(tokenizer.nextToken());
        context.emit(word.toLowerCase(), ONE);
    }
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(Comparable key, Double value, Context<Feature, Double> context) {
    MultiMap<Comparable, FeatureConfidenceTuple> multiMap = hazelcastInstance.getMultiMap("traindata-"+this.mapKey);

    Collection<FeatureConfidenceTuple> values = multiMap.get(key);

    for (FeatureConfidenceTuple featureConfidenceTuple : values) {
        context.emit(featureConfidenceTuple.getFeature(), (featureConfidenceTuple.getConfidenceCoefficient() * value.doubleValue()));
    }

}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(String key, Double value, Context<String, Double> context) {
    MultiMap<String, KeyValueTuple> multiMap = hazelcastInstance.getMultiMap("keyValueTuples");

    Collection<KeyValueTuple> values = multiMap.get(key);

    for (KeyValueTuple keyValueTuple : values) {
        context.emit(keyValueTuple.getKey(), (keyValueTuple.getValue()*value.doubleValue()) );
    }

}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(ByteHolder key, ClassifiedFeature value, Context<Map<String, Serializable>, Classification> context) {
    for (UnclassifiedFeature unclassifiedFeature : this.data) {
        List<Pair<Map<String, Serializable>, Classification>> list = new LinkedList<Pair<Map<String, Serializable>, Classification>>();
        Double coef = unclassifiedFeature.getConfidence() * value.getClassification().getConfidence();
        boolean similar = false;
        for (Map.Entry<String, Serializable> stringSerializableEntry : unclassifiedFeature.getFeatureMap().entrySet()) {
            for (Map.Entry<String, Serializable> stringSerializableEntryInner : value.getFeatureMap().entrySet()) {
                if (stringSerializableEntry.getValue().equals(stringSerializableEntryInner.getValue())) {
                    similar = true;
                } else {
                    Map<String, Serializable> otherKey = new HashMap<String, Serializable>();
                    otherKey.put("key", stringSerializableEntryInner.getValue());
                    list.add(new Pair<Map<String, Serializable>, Classification>(otherKey, new Classification(stringSerializableEntryInner.getValue(), coef)));
                }
            }
        }

        if (similar) {
            for (Pair<Map<String, Serializable>, Classification> pair : list) {
                context.emit(pair.fst, pair.snd);
            }
        }
        //System.out.println("New map:"+value.getClassification().toString());
    }

}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(Map<String, Serializable> key, Classification value, Context<Map<String, Serializable>, Classification> context) {
    for (UnclassifiedFeature unclassifiedFeature : this.data) {
        //double distance = unclassifiedFeature.getFeature().distanceTo(key);
        double distance = this.comparator.compare(unclassifiedFeature.getFeatureMap(), key);
        double weight = Double.MAX_VALUE;
        if (distance != 0) {
            weight = value.getConfidence() * unclassifiedFeature.getConfidence() / distance;
        }
        context.emit(unclassifiedFeature.getFeatureMap(), new Classification(value.getClassification(), weight));
        //System.out.println("New map:"+value.getClassification().toString());
    }

}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(String key, String numbersString, Context<Integer, Integer> context) {
    StringTokenizer tokenizer = new StringTokenizer(numbersString);

    while (tokenizer.hasMoreTokens()) {
        Integer number = Integer.valueOf(tokenizer.nextToken());
        context.emit(number, ONE);
    }
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override

    public void map(String key, String document, Context<String, Long> context) {
        // Just splitting the text by whitespaces
        StringTokenizer tokenizer = new StringTokenizer(document.toLowerCase());

        // For every token in the text (=> per word)
        while (tokenizer.hasMoreTokens()) {
            // Emit a new value in the mapped results
            context.emit(tokenizer.nextToken(), ONE);
        }
    }
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(String key, Integer value, Context<String, Integer> context) {
    AbstractMap.SimpleEntry entry = new AbstractMap.SimpleEntry(key, value);
    Integer valueOut = (Integer) supplier.apply((Map.Entry) entry);

    if (valueOut != null) {
        context.emit(key, valueOut);
    }
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(String key, Person value, Context<String, Person> context) {
    if (firstName.equalsIgnoreCase(value.getFirstName())) {
        context.emit(value.getEmail(), value);
    }
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(String key, SalaryYear value, Context<String, Integer> context) {
    context.emit("sum", value.getAnnualSalary());
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void initialize(final Context<Integer, Integer> context)
{
    m_insidePolygon = new InsidePolygon();
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void finalized(final Context<Integer, Integer> context)
{
}
 

import com.hazelcast.mapreduce.Context; //导入依赖的package包/类
@Override
public void map(Integer key, Employee employee, Context<Integer, Employee> context) {
    if (employee.getId() % mod == 0) {
        context.emit(key, employee);
    }
}