redis - Spark : How to send arguments to Spark foreach function

Question

我正在尝试使用以下代码将 Spark RDD 的内容保存到 Redis

import redis
class RedisStorageAdapter(BaseStorageAdapter):
    @staticmethod
    def save(record):

        ###--- How do I get action_name ---- ###
        redis_key = #<self.source_action_name>
        redis_host=settings['REDIS']['HOST']
        redis_port=settings['REDIS']['PORT']
        redis_db=settings['REDIS']['DB']

        redis_client = redis.StrictRedis(redis_host, redis_port, redis_db)
        redis_client.sadd(redis_key, record)

    def store_output(self, results_rdd):

        print self.source_action_name
        results_rdd.foreach(RedisStorageAdapter.save)

但我希望 Redis key 根据 self.source_action_name 初始化的内容(在 BaseStorageAdapter 中)而有所不同

如何将 source_action_name 传递给 RedisStorageAdapter.save 函数？ foreach函数只允许执行函数名，没有参数列表

此外 - 如果有更好的方法将数据从 RDD 移动到 Redis，请告诉我

Answer 1

当然，foreach 接受的是函数，而不是函数名。所以你可以向它传递一个 lambda 函数:

results_rdd.foreach(lambda x: RedisStorageAdapter.save(x, self.source_action_name))