提高 Apache Spark 到 Redis 的写入性能答案

【问题标题】：Improve Apache Spark to Redis writes performance提高 Apache Spark 到 Redis 的写入性能
【发布时间】：2019-05-08 06:57:25
【问题描述】：

我有一个使用 Apache Spark 将键值数据写入 Redis 的应用程序。该应用程序没有任何问题。但是，应用程序要慢得多。我在这里寻找一些建议，以提高写入吞吐量并在将数据写入 Redis 时增加并行度。

这里是代码

Dataset<Row> rowkeyMapping = services.select(regexp_replace(col("rowkey"), "_", "").as("rowkey"),struct(regexp_replace(col("name"), "\\[", ","), regexp_replace(col("oname"), "\\[", ","), col("cid")).as("detailsinfo"));

rowkeyMapping.foreach(obj -> {
    JedisPoolConfig poolConfig = new JedisPoolConfig();
    poolConfig.setMaxTotal(5000);
    JedisPool pool = new JedisPool(poolConfig, "redis-host", Integer.parseInt("6379"));
    Jedis jedis = pool.getResource();
    ObjectMapper om = new ObjectMapper();
    String[] rowArray = obj.mkString()
        .replaceAll("[\\[]", ",")
        .split(",");
    String key = rowArray[0];
    DetailInfo detail = new DetailInfo();
    detail.setName(rowArray[1]);
    detail.setOName(rowArray[2]);
    detail.setCid(rowArray[3]);

    String value = om.writeValueAsString(detail);
    logger.info("writing key value pairs to Redis cache (Key) :: " + key);
    jedis.set(key, value);
    jedis.quit();
});

我对 Redis Pipelining 了解甚少。但是，我认为流水线更多的是批处理命令。在这里，就我而言，我正在处理数百万个数据。我不确定流水线是否最合适。

任何帮助表示赞赏。在此先感谢。

【问题讨论】：

考虑使用github.com/RedisLabs/spark-redis库
好吧，我尝试使用这个库，但它有很多问题。于是，我开始使用 Jedis java 库。

标签： java apache-spark redis

【解决方案1】：

我不是 Spark 和 Redis 方面的专家，但我认为以下几行应该在 foreach 循环之外：

JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMaxTotal(5000);
JedisPool pool = new JedisPool(poolConfig, "redis-host", Integer.parseInt("6379"));

【讨论】：

问题是我们不能在foreach循环之外序列化连接。这就是为什么我必须用 foreach 循环实例化的原因。
为什么不能在foreach循环外序列化连接？
因为foreach是在executor中执行的。

【解决方案2】：

正如@Amir Kost 在他的回答中提到的，您的问题是您在设置单个键值对时创建了一个新连接。为了提高性能，您应该为一批键值对重用连接。

正如您在评论中提到的，您必须在执行程序中创建连接。所以为了复用连接，需要使用Dataset<Row>的foreachPartition方法，而不是foreach。 foreachPartition 为整个分区运行给定的 ForeachPartitionFunction<T> 函数。因此，您可以创建一个连接，并将其重用于分区中的所有项目。详情请查看doc。

另外，使用foreachPartition，可以获取分区中的一批项目，然后可以使用Redis Pipline获得更好的性能。查看管道doc了解详情

【讨论】：