2

我正在尝试使用以下代码将 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 密钥根据 self.source_action_name 的初始化内容而有所不同（在 BaseStorageAdapter 中）

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

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

4

1 回答 1

5

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

results_rdd.foreach(lambda x: RedisStorageAdapter.save(x, self.source_action_name))
于 2015-01-29T09:19:46.160 回答