spark-redis 异常：引起：redis.clients.jedis.exceptions.JedisConnectionException：java.net.SocketTimeoutException：读取超时

如何解决spark-redis 异常：引起：redis.clients.jedis.exceptions.JedisConnectionException：java.net.SocketTimeoutException：读取超时

我正在尝试通过 spark 将数据插入到 redis（用于 Redis 的 Azure 缓存）。大约有 7 亿行，我使用 spark-redis 连接器插入数据。它在抛出此错误后失败。我能够插入一些行，但一段时间后，一些任务开始失败并出现以下错误。我正在浏览 jupyter notebook。

Caused by: redis.clients.jedis.exceptions.JedisConnectionException: java.net.socketTimeoutException: Read timed out
    at redis.clients.jedis.util.RedisInputStream.ensureFill(RedisInputStream.java:205)
    at redis.clients.jedis.util.RedisInputStream.readByte(RedisInputStream.java:43)
    at redis.clients.jedis.Protocol.process(Protocol.java:155)
    at redis.clients.jedis.Protocol.read(Protocol.java:220)
    at redis.clients.jedis.Connection.readProtocolWithCheckingbroken(Connection.java:318)
    at redis.clients.jedis.Connection.getStatusCodeReply(Connection.java:236)
    at redis.clients.jedis.BinaryJedis.auth(BinaryJedis.java:2259)
    at redis.clients.jedis.JedisFactory.makeObject(JedisFactory.java:119)
    at org.apache.commons.pool2.impl.GenericObjectPool.create(GenericObjectPool.java:819)
    at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:429)
    at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:360)
    at redis.clients.jedis.util.Pool.getResource(Pool.java:50)
    ... 27 more
Caused by: java.net.socketTimeoutException: Read timed out
    at java.net.socketInputStream.socketRead0(Native Method)
    at java.net.socketInputStream.socketRead(SocketInputStream.java:116)
    at java.net.socketInputStream.read(SocketInputStream.java:171)
    at java.net.socketInputStream.read(SocketInputStream.java:141)
    at java.net.socketInputStream.read(SocketInputStream.java:127)
    at redis.clients.jedis.util.RedisInputStream.ensureFill(RedisInputStream.java:199)
    ... 38 more

这是我尝试写入数据的方式。

df.write
.option("host",REdis_URL)
.option("port",6379)
.option("auth",<PWD>)
.option("timeout",20000)
.format("org.apache.spark.sql.redis")
.option("table","testrediskeys").option("key.column","dummy").mode("overwrite").save()

Spark : 3.0
Scala : 2.12
spark-redis: com.redislabs:spark-redis_2.12:2.6.0

解决方法

我遇到了同样的问题，我的 spark 上下文的以下配置有所帮助：

val spark = SparkSession.builder()
      .appName("My-lovely-app")
      .master(options.masterSpec)
      .config("spark.redis.host",redisHost)
      .config("spark.redis.port",redisPort)
      .config("spark.redis.auth",redisPass)
      .config("spark.redis.timeout",redisSparkTimeout)
      .config("redis.timeout",redisTimeout)
      .config("spark.redis.max.pipeline.size",redisSparkMaxPipelineSize)
      .getOrCreate()

因此，您需要增加 spark.redis.timeout 和 redis.timeout 以获得更大的价值。两个配置的 3600000 毫秒（1 小时）值帮助我在 Redis Cluster 101 中加载了超过 5 亿个列表。对于大量加载（例如您的）优化，最好增加 spark.redis.max.pipeline.size。