import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
@Override
public void execute(HazelcastInstance hazelcastInstance)
        throws Exception {

    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    IList<Person> list = hazelcastInstance.getList("persons");
    KeyValueSource<String, Person> source = KeyValueSource.fromList(list);

    Job<String, Person> job = jobTracker.newJob(source);

    ICompletableFuture future = job.mapper(new SalaryMapper()) //
            .combiner(new SalaryCombinerFactory()) //
            .reducer(new SalaryReducerFactory()) //
            .submit();

    System.out.println(ToStringPrettyfier.toString(future.get()));
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
@Override
public void execute(HazelcastInstance hazelcastInstance)
        throws Exception {

    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    IMap<String, SalaryYear> map = hazelcastInstance.getMap("salaries");
    KeyValueSource<String, SalaryYear> source = KeyValueSource.fromMap(map);

    Job<String, SalaryYear> job = jobTracker.newJob(source);

    JobCompletableFuture<Integer> future = job //
            .mapper(new SalarySumMapper()) //
            .combiner(new SalarySumCombinerFactory()) //
            .reducer(new SalarySumReducerFactory()) //
            .submit(new SalarySumCollator());

    System.out.println("Salary sum: " + future.get());
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
@Override
public void execute(HazelcastInstance hazelcastInstance)
        throws Exception {

    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    IList<Person> list = hazelcastInstance.getList("persons");
    KeyValueSource<String, Person> source = KeyValueSource.fromList(list);

    Job<String, Person> job = jobTracker.newJob(source);

    // Collect all people by state
    ICompletableFuture future = job.mapper(new StateBasedCountMapper()).submit();

    // Count people by state
    // ICompletableFuture future = job.mapper(new StateBasedCountMapper()).reducer(new CountReducerFactory()).submit();

    // Same as above but with precalculation per node
    // ICompletableFuture future = job.mapper(new StateBasedCountMapper()).combiner(new CountCombinerFactory())
    //                                .reducer(new CountReducerFactory()).submit();

    System.out.println(ToStringPrettyfier.toString(future.get()));
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
public void execute(HazelcastInstance hazelcastInstance)
        throws Exception {

    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");
    IMap<String, Double> map = hazelcastInstance.getMap("inputs");

    map.put("EventA", new Double(Math.random() * 99 + 1));
    map.put("EventB", new Double(Math.random() * 99 + 1));

    KeyValueSource<String, Double> source = KeyValueSource.fromMap(map);

    Job<String, Double> job = jobTracker.newJob(source);

    JobCompletableFuture<List<KeyValueTuple>> future = job //
            .mapper(new InvertedMapper()) //
            .combiner(new InvertedCombinerFactory()) //
            .reducer(new InvertedReducerFactory()) //
            .submit(new InvertedCollator());

    System.out.println("Result: " + ToStringPrettyfier.toString(future.get()));
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
@Override
public Collection<Classification> predict(Collection<? extends Object> data) throws Exception {
    this.trainingdata = getTrainingdata();
    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    KeyValueSource<Map<String, Serializable>, Classification> source = KeyValueSource.fromMap(this.trainingdata);

    Job<Map<String, Serializable>, Classification> job = jobTracker.newJob(source);

    JobCompletableFuture<List<Classification>> future = job //
            .mapper(new DistanceBasedClassificationAlgorithmMapper(this.options,data)) //
            .combiner(new DistanceBasedClassificationAlgorithmCombinerFactory(this.options)) //
            .reducer(new DistanceBasedClassificationAlgorithmReducerFactory(this.options)) //
            .submit(new DistanceBasedClassificationAlgorithmCollator(this.options));

    return future.get();
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
@Override
    public Collection<Classification> predict(Collection<? extends Object> data) throws Exception {
        this.trainingdata = getTrainingdata();
        JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

        KeyValueSource<ByteHolder, ClassifiedFeature> source = KeyValueSource.fromMap(this.trainingdata);

        Job<ByteHolder, ClassifiedFeature> job = jobTracker.newJob(source);
//
        JobCompletableFuture<List<Classification>> future = job //
                .mapper(new UserBasedCollaborativeFilteringRecommendationAlgorithmMapper(this.options,data)) //
                .combiner(new DistanceBasedClassificationAlgorithmCombinerFactory(this.options)) //
                .reducer(new DistanceBasedClassificationAlgorithmReducerFactory(this.options)) //
                .submit(new DistanceBasedClassificationAlgorithmCollator(this.options));

        return future.get();
    }
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
private static double mapReduceAverage(HazelcastInstance hazelcastInstance)
        throws Exception {
    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");
    IMap<String, String> map = hazelcastInstance.getMap("numbers");
    KeyValueSource<String, String> source = KeyValueSource.fromMap(map);
    Job<String, String> job = jobTracker.newJob(source);

    ICompletableFuture<Double> future = job
            .mapper(new TokenizerMapper())
            .combiner(new NumberCountCombinerFactory())
            .reducer(new NumberCountAndOpReducerFactory())
            .submit(new AverageCollator());

    future.andThen(finishInfoCallback());

    return future.get();
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
private static Map<String, Long> mapReduce(HazelcastInstance hazelcastInstance)
        throws Exception {

    // Retrieving the JobTracker by name
    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    // Creating the KeyValueSource for a Hazelcast IMap
    IMap<String, String> map = hazelcastInstance.getMap("articles");
    KeyValueSource<String, String> source = KeyValueSource.fromMap(map);

    Job<String, String> job = jobTracker.newJob(source);

    // Creating a new Job
    ICompletableFuture<Map<String, Long>> future = job // returned future
            .mapper(new TokenizerMapper())             // adding a mapper
            .combiner(new WordCountCombinerFactory())  // adding a combiner through the factory
            .reducer(new WordCountReducerFactory())    // adding a reducer through the factory
            .submit();                                 // submit the task

    // Attach a callback listener
    future.andThen(buildCallback());

    // Wait and retrieve the result
    return future.get();
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
private static long mapReduceCollate(HazelcastInstance hazelcastInstance)
        throws Exception {

    // Retrieving the JobTracker by name
    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    // Creating the KeyValueSource for a Hazelcast IMap
    IMap<String, String> map = hazelcastInstance.getMap("articles");
    KeyValueSource<String, String> source = KeyValueSource.fromMap(map);

    // Creating a new Job
    Job<String, String> job = jobTracker.newJob(source);

    ICompletableFuture<Long> future = job // returned future
            .mapper(new TokenizerMapper())             // adding a mapper
            .combiner(new WordCountCombinerFactory())  // adding a combiner through the factory
            .reducer(new WordCountReducerFactory())    // adding a reducer through the factory
            .submit(new WordCountCollator());          // submit the task and supply a collator

    // Wait and retrieve the result
    return future.get();
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
@Override
public void execute(HazelcastInstance hazelcastInstance)
        throws Exception {

    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    IList<Person> list = hazelcastInstance.getList("persons");
    KeyValueSource<String, Person> source = KeyValueSource.fromList(list);

    Job<String, Person> job = jobTracker.newJob(source);

    ICompletableFuture<List<Map.Entry<String, Integer>>> future = //
            job.mapper(new SalaryMapper()) //
                    .combiner(new SalaryCombinerFactory()) //
                    .reducer(new SalaryReducerFactory()) //
                    .submit(new SalaryCollator());

    // Intermediate result
    List<Map.Entry<String, Integer>> orderedSalariesByState = future.get();
    Map.Entry<String, Integer> topSalary = orderedSalariesByState.get(0);

    IList<Crime> crimesList = hazelcastInstance.getList("crimes");
    KeyValueSource<String, Crime> crimeSource = KeyValueSource.fromList(crimesList);

    Job<String, Crime> crimeJob = jobTracker.newJob(crimeSource);

    ICompletableFuture<Map<CrimeCategory, Integer>> crimeFuture = //
            crimeJob.mapper(new CrimeMapper(topSalary.getKey())) //
                    .reducer(new CrimeReducerFactory()) //
                    .submit();

    System.out.println(ToStringPrettyfier.toString(crimeFuture.get()));
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
public static void main(String[] args)
        throws Exception {

    // Prepare Hazelcast cluster
    HazelcastInstance hazelcastInstance = buildCluster(3);

    try {

        // Read data
        fillMapWithData(hazelcastInstance);

        JobTracker tracker = hazelcastInstance.getJobTracker("default");

        IMap<String, String> map = hazelcastInstance.getMap(MAP_NAME);
        KeyValueSource<String, String> source = KeyValueSource.fromMap(map);

        Job<String, String> job = tracker.newJob(source);
        ICompletableFuture<Map<String, Integer>> future = job
                .mapper(new TokenizerMapper())
                // Activate Combiner to add combining phase!
                // .combiner(new WordcountCombinerFactory())
                .reducer(new WordcountReducerFactory())
                .submit();



        System.out.println(ToStringPrettyfier.toString(future.get()));

    } finally {
        // Shutdown cluster
        Hazelcast.shutdownAll();
    }
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
private static void doMapReduce(
        final HazelcastInstance instance,
        final String filename) throws ExecutionException, InterruptedException, IOException
{
    final IMap<Integer, SpatialPoint> geomap = instance.getMap(GEOMAP);
    final JobTracker tracker = instance.getJobTracker("default");
    final Job<Integer, SpatialPoint> job = tracker.newJob(KeyValueSource.fromMap(geomap));
    final ICompletableFuture<Map<Integer, Integer>> future = job.
            mapper(new FeatureMapper(HEXMAP)).
            reducer(new FeatureReducerFactory()).
            submit();

    final FileOutputStream fileOutputStream = new FileOutputStream(filename);
    try
    {
        final PrintStream printStream = new PrintStream(fileOutputStream);
        printStream.format("ID,POPULATION%n");
        final Map<Integer, Integer> result = future.get();
        for (final Map.Entry<Integer, Integer> entry : result.entrySet())
        {
            printStream.format("%d,%d%n", entry.getKey(), entry.getValue());
        }
        printStream.flush();
    }
    finally
    {
        fileOutputStream.close();
    }
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
@TimeStep(prob = 0.5)
public void mapReduce(ThreadState state) throws Exception {
    JobTracker tracker = targetInstance.getJobTracker(Thread.currentThread().getName() + name);
    KeyValueSource<Integer, Employee> source = KeyValueSource.fromMap(map);
    Job<Integer, Employee> job = tracker.newJob(source);

    ICompletableFuture<Map<Integer, Set<Employee>>> future = job
            .mapper(new ModIdMapper(2))
            .combiner(new RangeIdCombinerFactory(10, 30))
            .reducer(new IdReducerFactory(10, 20, 30))
            .submit();

    Map<Integer, Set<Employee>> result = future.get();

    for (Set<Employee> set : result.values()) {
        for (Employee employee : set) {

            assertTrue(employee.getId() % 2 == 0);
            assertTrue(employee.getId() >= 10 && employee.getId() <= 30);
            assertTrue(employee.getId() != 10);
            assertTrue(employee.getId() != 20);
            assertTrue(employee.getId() != 30);
        }
    }

    state.operationCounter.mapReduce++;
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
public static void main(String[] args)
    throws Exception {

    // Prepare Hazelcast cluster
    HazelcastInstance hazelcastInstance = buildCluster(3);

    try {

        // Read data
        fillMapWithData(hazelcastInstance);

        JobTracker tracker = hazelcastInstance.getJobTracker(TRACKER_NAME);

        IMap<String, String> map = hazelcastInstance.getMap(MAP_NAME);
        KeyValueSource<String, String> source = KeyValueSource.fromMap(map);

        Job<String, String> job = tracker.newJob(source);

        final JobCompletableFuture<List<Map.Entry<String, Integer>>> future = job
            .mapper(new TokenizerMapper())
            // Activate Combiner to add combining phase!
            // .combiner(new WordcountCombinerFactory())
            .reducer(new WordcountReducerFactory())
            //                .submit();
            // add collator for sorting and top10
            .submit(new WordcountCollator());

        future.andThen(new ExecutionCallback<List<Map.Entry<String, Integer>>>() {
            @Override public void onResponse(List<Map.Entry<String, Integer>> response) {
                System.out.println(ToStringPrettyfier.toString(response));
            }

            @Override public void onFailure(Throwable t) {

            }
        });

        //System.out.println(ToStringPrettyfier.toString(future.get()));

    } finally {
        // Shutdown cluster
        //Hazelcast.shutdownAll();
    }
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
@Override
public void execute(HazelcastInstance hazelcastInstance)
        throws Exception {

    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    IList<Person> list = hazelcastInstance.getList("persons");
    KeyValueSource<String, Person> source = KeyValueSource.fromList(list);

    Job<String, Person> job = jobTracker.newJob(source);

    // Find all people named James
    ICompletableFuture future = job.mapper(new PersonMapper("James")).submit();

    System.out.println(ToStringPrettyfier.toString(future.get()));
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
@Override
public void execute(HazelcastInstance hazelcastInstance)
        throws Exception {

    JobTracker jobTracker = hazelcastInstance.getJobTracker("default");

    IList<Person> list = hazelcastInstance.getList("persons");
    KeyValueSource<String, Person> source = KeyValueSource.fromList(list);

    Job<String, Person> job = jobTracker.newJob(source);

    // Find all people grouped by state
    // ICompletableFuture future = job.mapper(new StateBasedMapper()).submit();

    // Find all people for the given state
    ICompletableFuture future = job.mapper(new StateBasedMapper("CA")).submit();

    System.out.println(ToStringPrettyfier.toString(future.get()));
}
 

import com.hazelcast.mapreduce.KeyValueSource; //导入依赖的package包/类
@Test
public void testMR() throws IOException, ExecutionException, InterruptedException
{
    final Config config = new ClasspathXmlConfig("hazelcast.xml");
    final HazelcastInstance instance = Hazelcast.newHazelcastInstance(config);
    try
    {
        final IMap<Integer, SpatialPolygon> hexmap = instance.getMap("hexmap");

        final SpatialPolygon spatialPolygon = new SpatialPolygon();
        spatialPolygon.id = 0;
        spatialPolygon.polygon.startPath(0, 0);
        spatialPolygon.polygon.lineTo(10, 0);
        spatialPolygon.polygon.lineTo(10, 10);
        spatialPolygon.polygon.lineTo(0, 10);
        spatialPolygon.polygon.lineTo(0, 0);
        spatialPolygon.polygon.closeAllPaths();
        spatialPolygon.polygon.queryEnvelope2D(spatialPolygon.envelope2D);
        hexmap.put(spatialPolygon.id, spatialPolygon);

        final IMap<Integer, SpatialPoint> geomap = instance.getMap("geomap");

        final SpatialPoint spatialPoint = new SpatialPoint();
        spatialPoint.id = 0;
        spatialPoint.point.setXY(5, 5);
        spatialPoint.point.queryEnvelope2D(spatialPoint.envelope2D);
        geomap.put(spatialPoint.id, spatialPoint);

        final JobTracker tracker = instance.getJobTracker("default");
        final Job<Integer, SpatialPoint> job = tracker.newJob(KeyValueSource.fromMap(geomap));
        final ICompletableFuture<Map<Integer, Integer>> future = job.
                mapper(new FeatureMapper("hexmap")).
                reducer(new FeatureReducerFactory()).
                submit();

        final Map<Integer, Integer> result = future.get();
        Assert.assertEquals(1, result.size());
        final Map.Entry<Integer, Integer> next = result.entrySet().iterator().next();
        Assert.assertEquals(0, next.getKey().intValue());
        Assert.assertEquals(1, next.getValue().intValue());
    }
    finally
    {
        instance.shutdown();
    }
}