赞
踩
利用 Spark 分布式集群强悍能力,实现高 QPS 写入 Redis 能力,QPS 在一定范围内支持线性扩展。注意解决 Redis Pool 不能序列化问题。
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
import redis.clients.jedis.JedisPoolConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; public class RedisPool { private static final Logger log = LoggerFactory.getLogger(RedisPool.class); private volatile static JedisPool pool; public static void init(String host, int port, String password, int size) { if(pool == null) { synchronized (RedisPool.class) { if (pool == null) { JedisPoolConfig jedisPoolConfig = new JedisPoolConfig(); jedisPoolConfig.setMaxIdle(10); jedisPoolConfig.setMaxTotal(20); jedisPoolConfig.setMinIdle(5); jedisPoolConfig.setMaxWaitMillis(10 * 1000); jedisPoolConfig.setTestWhileIdle(true); jedisPoolConfig.setTestOnBorrow(false); jedisPoolConfig.setTestOnReturn(false); pool = new JedisPool(jedisPoolConfig, host, port, 10 * 1000, password); } } } } /** * 为指定的 k 设置值及过期时间。如果 k 已经存在,setex 命令会替换旧值和过期时间 * * @param k * @param seconds * @param v * @return * @throws Exception */ public static String setex(String k, int seconds, String v) throws Exception { Jedis jedis = null; try { jedis = pool.getResource(); return jedis.setex(k, seconds, v); } catch (Exception e) { String errInfo = String.format("[RedisPool]setex failed, k=%s, v=%s, seconds=%d, err=%s", k, v, seconds, e.getMessage()); log.error(errInfo); throw e; } finally { if(jedis != null) { jedis.close(); } } } /** * 用于获取指定 k 的值。如果 k 不存在,返回 null * * @param k * @return */ public static String get(String k) throws Exception { Jedis jedis = null; try { jedis = pool.getResource(); return jedis.get(k); } catch (Exception e) { String errInfo = String.format("[RedisPool]get failed, k=%s, err=%s", k, e.getMessage()); log.error(errInfo); throw e; } finally { if(jedis != null) { jedis.close(); } } } /** * 设置 key 的过期时间 * * @param k * @param seconds * @return */ public static Long expire(String k, Integer seconds){ Jedis jedis = null; try { jedis = pool.getResource(); return jedis.expire(k, seconds); } catch (Exception e) { String errInfo = String.format("[RedisPool]expire failed, k=%s, seconds=%d, err=%s", k, seconds, e.getMessage()); log.error(errInfo); throw e; } finally { if(jedis != null) { jedis.close(); } } } /** * 返回 key 的剩余过期时间,单位是秒 * * @param k * @return */ public static Long ttl(String k){ Jedis jedis = null; try { jedis = pool.getResource(); return jedis.ttl(k); } catch (Exception e) { String errInfo = String.format("[RedisPool]ttl failed, k=%s, err=%s", k, e.getMessage()); log.error(errInfo); throw e; } finally { if(jedis != null) { jedis.close(); } } } }
val df = spark.read.parquet(path) df .repartition(partitions) // partitions 作为参数传入,便于自由调节并行能力 .as[CaseClass] .foreachPartition { its => // 各个 Executor 上初始化 Redis Pool,解决序列化问题 RedisPool.init(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD, REDIS_CONNECTION_SIZE) its.foreach { x => val k = s"$REDIS_KEY_PREFIX:${x.k}" val v = JsonUtils.toJson(x) RedisPool.setex(k, REDIS_KEY_TTL, v) } }
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。