Spark：如何将参数发送到 Spark foreach 函数

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)

但我希望根据 self.source_action_name 初始化的内容（在 BaseStorageAdapter 中）使 Redis 密钥不同

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

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

Answer 1

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

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

Spark：如何将参数发送到 Spark foreach 函数

Spark: How to send arguments to Spark foreach function

redis

apache-spark

pyspark