11.4　Spark与Redis整合原理与实战

本节通过生产环境实战案例讲解Spark与Redis整合原理及Spark与Redis整合实战。

11.4.1　Spark与Redis整合原理

Redis是一个开源项目（BSD许可）。Redis以内存数据结构存储，用作数据库、缓存和消息代理。它支持数据结构，如字符串、散列、列表、集合、具有范围查询的排序集、位图、超文本和具有半径查询的地理空间索引等。Redis内置复制、Lua脚本、LRU eviction、事务和不同级别的磁盘持久性，通过Redis Sentinel提供高可用性，并通过Redis Cluster进行自动分区。

Redis可以对这些类型运行原子操作，如附加到字符串、在哈希中增加值、将元素推送到列表中、计算集交集、联合与差异于一体；或者在排序集中获得最高排名的成员。

为了实现其卓越的性能，Redis使用内存中的数据集。根据业务用例，可以通过将数据集一次性转储到磁盘中，或通过将每个命令附加到日志来持久化。如果只需要功能丰富的网络内存缓存，则可以选择禁用持久性。

Redis还支持简单的主从异步复制，第一次同步非阻塞的速度非常快，网络切分传输时可自动重连同步。

Redis的其他功能包括：

　事务Transactions。

　发布/订阅。

　Lua脚本。

　Keys with a limited time-to-live。

　LRU eviction of keys。

　自动故障切换。

大多数编程语言可以使用Redis。

Redis以ANSI C编写，适用于大多数POSIX系统，如Linux、* BSD、OS X，无需外部依赖。Linux和OS X是Redis开发和测试的两个操作系统，建议使用Linux进行部署。Redis可能在诸如SmartOS的Solaris衍生系统中工作，但支持是尽力而为的。没有官方支持Windows版本，但是Microsoft开发并维护了Redis的Win-64端口。

11.4.2　Spark与Redis整合实战

本节以生产环境中Spark与Redis整合实战案例来讲解。在通信运营商的Spark大数据项目中，Spark每分钟实时读取Hdfs中的话单数据，经过业务逻辑代码分析转换后，将清洗以后的数据转换成一个List字符串列表，然后遍历List字符串列表，将每条记录拼接成一个Key-Value字符串放入Redis队列。

Spark与Redis整合实战案例实现步骤如下。

（1）通过Maven方式下载Redis的jedis 2.6.0 Jar包。

pom.xml文件增加以下内容：

1.  <dependency>
2.        <groupId>redis.clients</groupId>
3.        <artifactId>jedis</artifactId>
4.        <version>2.6.0</version>
5.    </dependency>

在Spark中导入Redis的Jar包。

1.  import redis.clients.jedis.Jedis;
2.  import redis.clients.jedis.JedisPool;
3.  import redis.clients.jedis.JedisPoolConfig;

（2）在项目config.properties配置文件中增加Redis的主机地址、端口、密码，以及Redis连接池分配的连接数、等待时间等信息。

1.  ## REDIS
2.  redis.ip=100.*.*.100
3.  redis.port=6379
4.  redis.password=password
5.  ......
6.  ## redis
7.  #最大分配的对象数
8.  redis.pool.maxTotal=1024
9.  #最大能够保持idle状态的对象数
10. redis.pool.maxIdle=200
11. #当池内没有返回对象时，最大等待时间
12. redis.pool.maxWait=1000
13. #当调用borrow Object方法时，是否进行有效性检查
14. redis.pool.testOnBorrow=true
15. #当调用return Object方法时，是否进行有效性检查
16. redis.pool.testOnReturn=true
17.

（3）编写RedisServiceImpl实现类，从配置文件中获取Redis的主机地址、端口、密码等信息；编写getFromPool方法从redis访问池中获取redis实例；编写getSingle方法获取redis实例。

1.   public class RedisServiceImpl {
2.      private static final Map<String, String> REDIS_CONFIG = Config.
        getInstance().getRedisParams();
3.      private static final String REDIS_IP = REDIS_CONFIG.get("redis.ip");
4.      private static final String REDIS_PORT = REDIS_CONFIG.get("redis.port");
5.      private static final String REDIS_PASSWORD = REDIS_CONFIG.get("redis.
        password");
6.      private static final int REDIS_TIMEOUT = 2000;
7.      private static Logger log = LoggerFactory.getLogger(RedisServiceImpl.class);
8.      private static JedisPool pool;
9.      /**
          * 从redis访问池中获取redis实例
          *
          * @return redis实例
10.       */
11.     public static Jedis getFromPool() {
12.         if (pool == null) {
13.             JedisPoolConfig config = new JedisPoolConfig();
14.             config.setMaxTotal(Integer.valueOf(REDIS_CONFIG.get("redis.
                 pool.maxTotal")));
15.             config.setMaxIdle(Integer.valueOf(REDIS_CONFIG.get("redis.
                 pool.maxIdle")));
16.             config.setMaxWaitMillis(Integer.valueOf(REDIS_CONFIG.get("redis.
                 pool.maxWait")));
17.             config.setTestOnBorrow(Boolean.valueOf(REDIS_CONFIG.get("redis.
                 pool.testOnBorrow")));
18.             config.setTestOnReturn(Boolean.valueOf(REDIS_CONFIG.get("redis.
                 pool.testOnReturn")));
19.             pool = new JedisPool(config, REDIS_IP, Integer.valueOf(REDIS_
                PORT), REDIS_TIMEOUT, REDIS_PASSWORD);
20.         }
21.         return pool.getResource();
22.     }
23.
24.     public static void returnResource(Jedis redis) {
25.         pool.returnResource(redis);
26.     }
27.
28.     /**
          * 获取redis实例
          *
          * @return redis实例
29.       */
30.
31.     public static Jedis getSingle() {
32.         Jedis redis = new Jedis(REDIS_IP, Integer.valueOf(REDIS_PORT),
            REDIS_TIMEOUT);
33.         redis.auth(REDIS_PASSWORD);
34.         return redis;
35.     }
36. }

（4）编写项目的业务代码，RedisBean类数据结构用于要存放Redis的数据。

1.  public class RedisBean {
2.      public void setKey(String key) {
3.         this.key = key;
4.     }
5.
6.     public void setValue(String value) {
7.         this.value = value;
8.     }
9.
10.    public String getKey() {
11.        return key;
12.    }
13.
14.    public String getValue() {
15.        return value;
16.    }
17.
18.    protected String key;
19.    protected String value;
20.
21. }

根据项目的业务需求，将Spark中提取转换后的每条记录存入业务数据结构RedisBean的key、value中；然后加入到RedisBeanList列表中。RedisBeanList是List<RedisBean> RedisBeanList类型。

1.  if ("RedisTestQUALINFO".equals(keyType)) {
2.         RedisBean.key = "RedisTestQUALINFO";
3.         RedisBean.value = RedisTestReslut.toString();
4.         RedisBeanList.add(RedisBean);
5.     }

（5）最终调用业务方法addToRedis，通过RedisServiceImpl.getSingle()获取Redis实例，然后循环遍历List<RedisBean>的每个元素，调用redis.lpush方法分别将Key值、Value值lpush到Redis中。lPush完成以后，通过redis.close关闭连接。

1.     public static void addToRedis(List<RedisBean> redisBeanList) {
2.     Jedis redis = RedisServiceImpl.getSingle();
3.     for (RedisBean redisData : redisBeanList) {
4.         redis.lpush(redisData.getKey(), redisData.getValue());
5.     }
6.     redis.close();
7.  }

（6）Redis业务验证：可以登录到Redis系统中，查询数据已持久化至Redis。

1.   redis-cli –h 100.*.*.100  -p 6379 -a 'password'
2.  select 0  ---切换到0库
3.  keys *    ---列出所有的key
4.  lrange CDNNODEQUALINFO 0 -1 查看所有的记录
本周热推：
微机原理及接口技术网络安全概论Project 2007项目管理实用详解电脑办公嵌入式系统及其开发应用
上一章目录下一章