主要包括redis連接池，重寫FileOutputFormat函數(shù)。

redis連接池

/**
* redis 連接池
*/
public class RedisHelper {

    private static JedisPool jedisPool;

    static {
        init();
    }

    public synchronized static Jedis getJedis() {
        if (jedisPool != null) {
            Jedis resource = jedisPool.getResource();
            return resource;
        } else {
            return null;
        }
    }

    /**
     * 釋放jedis資源
     *
     * @param jedis
     */
    public static void close(final Jedis jedis) {
        if (jedis != null) {
            jedis.close();
        }
    }

    public static void destroy() {
        if (jedisPool != null) {
            jedisPool.destroy();
        }
    }

public static void putValue(Jedis jedis, String key, String value) {
        jedis.set(key, value);
    }

    private static void init() {
        Properties prop = new Properties();
        try (InputStream in = RedisHelper.class.getResourceAsStream("/database_config.properties")) {
            prop.load(in);
            String redisHost = prop.getProperty("redisHost");
            int redisPort = Integer.valueOf(prop.getProperty("redisPort"));
            String redisPassword = prop.getProperty("redisPassword");
            JedisPoolConfig config = new JedisPoolConfig();
            config.setMaxTotal(Integer.valueOf(prop.getProperty("redisMaxTotal")));
            config.setMaxIdle(Integer.valueOf(prop.getProperty("redisMaxIdle")));
            config.setMinIdle(Integer.valueOf(prop.getProperty("redisMinIdle")));
            config.setMaxWaitMillis(Long.valueOf(prop.getProperty("redisMaxWaitMillis")));
            config.setTestOnBorrow(Boolean.parseBoolean(prop.getProperty("redisTestOnBorrow")));

            jedisPool = new JedisPool(config, redisHost, redisPort, 2000, redisPassword);
        } catch (IOException e) {
            log.error("Failed to initialize redis parameters. ", e);
            exit(1);
        }
    }

    private RedisHelper() {
    }
}

重寫FileOutputFormat函數(shù)

public class ResultOutputFormat<K, V> extends FileOutputFormat<K, V> {

    /**
     * 定制一個(gè)RecordWriter類，每一條reduce處理后的記錄，將該記錄輸出到數(shù)據(jù)庫中
     */
    protected static class RedisRecordWriter<K, V> extends RecordWriter<K, V> {

        private Jedis jedis;

        public RedisRecordWriter(Jedis jedis) {
            this.jedis = jedis;
        }

        @Override
        public void write(K key, V value) throws IOException, InterruptedException {
            if (key == null || value == null) {
                return;
            }
            RedisHelper.putValue(jedis, key.toString(), value.toString());
        }

        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            RedisHelper.close(jedis); //關(guān)閉鏈接
        }
    }

    @Override
    public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
        Jedis jedis = RedisHelper.getJedis();
        return new RedisRecordWriter<K, V>(jedis);
    }
}

然后在job設(shè)置處寫入
job.setOutputFormatClass(ResultOutputFormat.class);

完美~~