Redis в Spark: задача не сериализуема

В Spark функции на RDDs (например, map здесь) сериализуются и отправляются исполнителям для обработки. Это означает, что все элементы, содержащиеся в этих операциях, должны быть сериализуемыми.

Соединение Redis здесь не сериализуемо, поскольку оно открывает TCP-соединения с целевой БД, которые привязаны к машине, на которой оно создано.

Решение состоит в том, чтобы создать эти соединения на исполнителях в локальном контексте выполнения. Есть несколько способов сделать это. Два, которые приходят на ум:

• rdd.mapPartitions: позволяет обрабатывать сразу весь раздел и, следовательно, амортизировать стоимость создания подключений)
• Диспетчеры соединений Singleton: создайте соединение один раз для каждого исполнителя.

mapPartitions проще, так как для этого требуется лишь небольшое изменение в структуре программы:

val perhit = perhitFile.mapPartitions{partition => 
    val r = new RedisClient("192.168.1.101", 6379) // create the connection in the context of the mapPartition operation
    val res = partition.map{ x =>
        ...
        val refStr = r.hmget(...) // use r to process the local data
    }
    r.close // take care of resources
    res
}

Диспетчер одноэлементных соединений можно смоделировать с помощью объекта, который содержит ленивую ссылку на соединение (примечание: изменяемая ссылка также будет работать).

object RedisConnection extends Serializable {
   lazy val conn: RedisClient = new RedisClient("192.168.1.101", 6379)
}

Затем этот объект можно использовать для создания экземпляра 1 соединения для каждой рабочей JVM и использовать как объект Serializable при закрытии операции.

val perhit = perhitFile.map{x => 
    val param = f(x)
    val refStr = RedisConnection.conn.hmget(...) // use RedisConnection to get a connection to the local data
    }
}

Преимущество использования одноэлементного объекта заключается в меньших накладных расходах, поскольку JVM создает соединения только один раз (в отличие от одного на раздел RDD).

Есть и недостатки:

• очистка соединений сложна (крючок / таймеры выключения)
• необходимо обеспечить потокобезопасность общих ресурсов

(*) код предоставлен для иллюстрации. Не компилируется и не тестируется.

Вы пытаетесь сериализовать клиент. У вас есть один RedisClient, r, который вы пытаетесь использовать внутри map, который будет выполняться на разных узлах кластера. Либо получите нужные данные из Redis отдельно перед выполнением задачи кластера, либо создайте клиент отдельно для каждой задачи кластера внутри вашего блока map (возможно, используя mapPartitions, а не map, как создание нового Redis). клиент для каждой отдельной строки, вероятно, плохая идея).