Hazelcast 简明教程

Hazelcast - Map Reduce & Aggregations

MapReduce 是一种计算模型，在您有大量数据且需要多台机器（即分布式环境）计算数据时，该模型对数据处理非常有用。它涉及将数据映射为键值对，然后进行“缩减”，即对这些键进行分组并在值上执行操作。

MapReduce is a computation model which is useful for data processing when you have lots of data and you need multiple machines, i.e., a distributed environment to calculate data. It involves 'map’ing of data into key-value pairs and then 'reducing', i.e., grouping these keys and performing operation on the value.

鉴于 Hazelcast 是在考虑分布式环境的情况下设计的，因此它自然而然地实现了 Map-Reduce 框架。

Given the fact that Hazelcast is designed keeping a distributed environment in mind, implementing Map-Reduce Frameworks comes naturally to it.

让我们用一个例子看看该怎么做。

Let’s see how to do it with an example.

例如，让我们假设我们有关于汽车（品牌和车号）及其车主的数据。

For example, let’s suppose we have data about a car (brand & car number) and the owner of that car.

Honda-9235, John
Hyundai-235, Alice
Honda-935, Bob
Mercedes-235, Janice
Honda-925, Catnis
Hyundai-1925, Jane

现在，我们必须找出每个品牌的汽车数量，即现代、本田等。

And now, we have to figure out the number of cars for each brand, i.e., Hyundai, Honda, etc.

Example

让我们尝试使用 MapReduce 找出 −

Let’s try to find that out using MapReduce −

package com.example.demo;

import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ICompletableFuture;
import com.hazelcast.core.IMap;
import com.hazelcast.mapreduce.Context;
import com.hazelcast.mapreduce.Job;
import com.hazelcast.mapreduce.JobTracker;
import com.hazelcast.mapreduce.KeyValueSource;
import com.hazelcast.mapreduce.Mapper;
import com.hazelcast.mapreduce.Reducer;
import com.hazelcast.mapreduce.ReducerFactory;

public class MapReduce {
   public static void main(String[] args) throws ExecutionException,
   InterruptedException {
      try {
         // create two Hazelcast instances
         HazelcastInstance hzMember = Hazelcast.newHazelcastInstance();
         Hazelcast.newHazelcastInstance();
         IMap<String, String> vehicleOwnerMap=hzMember.getMap("vehicleOwnerMap");
         vehicleOwnerMap.put("Honda-9235", "John");
         vehicleOwnerMap.putc"Hyundai-235", "Alice");
         vehicleOwnerMap.put("Honda-935", "Bob");
         vehicleOwnerMap.put("Mercedes-235", "Janice");
         vehicleOwnerMap.put("Honda-925", "Catnis");
         vehicleOwnerMap.put("Hyundai-1925", "Jane");
         KeyValueSource<String, String> kvs=KeyValueSource.fromMap(vehicleOwnerMap);
         JobTracker tracker = hzMember.getJobTracker("vehicleBrandJob");
         Job<String, String> job = tracker.newJob(kvs);
         ICompletableFuture<Map<String, Integer>> myMapReduceFuture =
            job.mapper(new BrandMapper())
            .reducer(new BrandReducerFactory()).submit();
         Map<String, Integer&g; result = myMapReduceFuture.get();
         System.out.println("Final output: " + result);
      } finally {
         Hazelcast.shutdownAll();
      }
   }
   private static class BrandMapper implements Mapper<String, String, String, Integer> {
      @Override
      public void map(String key, String value, Context<String, Integer>
      context) {
         context.emit(key.split("-", 0)[0], 1);
      }
   }
   private static class BrandReducerFactory implements ReducerFactory<String, Integer, Integer> {
      @Override
      public Reducer<Integer, Integer> newReducer(String key) {
         return new BrandReducer();
      }
   }
   private static class BrandReducer extends Reducer<Integer, Integer> {
      private AtomicInteger count = new AtomicInteger(0);
      @Override
      public void reduce(Integer value) {
         count.addAndGet(value);
      }
      @Override
      public Integer finalizeReduce() {
         return count.get();
      }
   }
}

让我们尝试理解此代码 −

Let’s try to understand this code −

  1. We create Hazelcast members. In the example, we have a single member, but there can well be multiple members.

  2. We create a map using dummy data and create a Key-Value store out of it.

  3. We create a Map-Reduce job and ask it to use the Key-Value store as the data.

  4. We then submit the job to cluster and wait for completion.

  5. The mapper creates a key, i.e., extracts brand information from the original key and sets the value to 1 and then emits that information as K-V to the reducer.

  6. The reducer simply sums the value, grouping the data, based on key, i.e., brand name.

Output

代码的输出 −

The output of the code −

Final output: {Mercedes=1, Hyundai=2, Honda=3}