1

在下面的代码中，我尝试使用 URL 上的 env 变量实例化redis-py 连接。问题是，当我使用foreach 或 foreachPartition时，#save_on_redis 方法无法识别 env 变量。

我只是尝试在外部创建 redis 连接，但收到"pickle.PicklingError: Can't pickle 'lock' object"，因为 spark 尝试在所有节点上同时运行这两种方法。

问题：如何在作为参数传递给 foreach 或 foreachPartition 的方法上使用环境变量？

import os
from pyspark.sql import SparkSession
import redis

spark = (SparkSession
        .builder
        .getOrCreate())

print "---------"
print os.getenv("REDIS_REPORTS_URL")
print "---------"

def save_on_redis(row):
    redis_ = redis.StrictRedis(host=os.getenv("REDIS_REPORTS_URL"), port=6379, db=0)
    print os.getenv("REDIS_REPORTS_URL")
    print redis_
    redis_.set("#teste#", "fagner")


df  = spark.createDataFrame([(0,1), (0,1), (0,2)], ["id", "score"])
df.foreach(save_on_redis)
4

1 回答 1

1

我建议您在驱动程序进程中获取 env 变量并将其作为 python 变量传递给工作进程，您可以在其中使用设置您的环境os.putenv

例子：

In [1]: import os

In [2]: a = sc.parallelize(range(20))

In [3]: os.getenv('MY_VAR')
Out[3]: 'some_value'

In [4]: def f(iter):
    import os
    return (str(os.getenv('MY_VAR')),)
   ...:

In [5]: a.mapPartitions(f).collect()
Out[5]: ['None', 'None']

In [6]: my_var = os.getenv('MY_VAR')

In [6]: def f2(iter):
    import os
    from subprocess import check_output
    os.putenv('MY_VAR', my_var)
    return (check_output('env | grep MY_VAR', shell=True), my_var)
   ....:

In [7]: a.mapPartitions(f2).collect()
Out[7]:
['MY_VAR=some_value\n',
 'some_value',
 'MY_VAR=some_value\n',
 'some_value']

PS。根据这个答案，最好直接修改os.environ映射对象而不是使用os.putenv

于 2017-05-11T13:45:15.060 回答