python-3.x - 如何在 AWS EMR 上设置 PYTHONHASHSEED

Question

有没有办法在 EMR 集群的所有节点上设置环境变量？

尝试在 Python3 PySpark 中使用 reduceByKey() 时遇到错误，并收到有关哈希种子的错误。我可以看到这是一个已知错误，并且需要在集群的所有节点上将环境变量 PYTHONHASHSEED 设置为相同的值，但我对此没有任何运气。

我尝试通过集群配置向 spark-env 添加一个变量:

[
  {
    "Classification": "spark-env",

      "Configurations": [
      {
         "Classification": "export",
         "Properties": {
          "PYSPARK_PYTHON": "/usr/bin/python3",
          "PYTHONHASHSEED": "123"
       }
     }
   ]
 },
 {
   "Classification": "spark",
   "Properties": {
     "maximizeResourceAllocation": "true"
    }
  }
]

但这不起作用。我还尝试添加引导脚本:

#!/bin/bash
export PYTHONHASHSEED=123

但这似乎也不起作用。

Answer 1

我相信 /usr/bin/python3 没有选择您在 PYTHONHASHSEED 范围下的集群配置中定义的环境变量 spark-env 。

您应该使用 python34 而不是 /usr/bin/python3 并按如下方式设置配置:

[
   {
      "classification":"spark-defaults",
      "properties":{
         // [...]
      }
   },
   {
      "configurations":[
         {
            "classification":"export",
            "properties":{
               "PYSPARK_PYTHON":"python34",
               "PYTHONHASHSEED":"123"
            }
         }
      ],
      "classification":"spark-env",
      "properties":{
        // [...]
      }
   }
]

现在，让我们测试一下。我定义了一个 bash 脚本调用 python s:

#!/bin/bash

echo "using python34"
for i in `seq 1 10`;
  do
    python -c "print(hash('foo'))";
  done
echo "----------------------"
echo "using /usr/bin/python3"
for i in `seq 1 10`;
  do
    /usr/bin/python3 -c "print(hash('foo'))";
  done

判决 :

[hadoop@ip-10-0-2-182 ~]$ bash test.sh
using python34
-4177197833195190597
-4177197833195190597
-4177197833195190597
-4177197833195190597
-4177197833195190597
-4177197833195190597
-4177197833195190597
-4177197833195190597
-4177197833195190597
-4177197833195190597
----------------------
using /usr/bin/python3
8867846273747294950
-7610044127871105351
6756286456855631480
-4541503224938367706
7326699722121877093
3336202789104553110
3462714165845110404
-5390125375246848302
-7753272571662122146
8018968546238984314

PS1: 我正在使用 AMI 版本 emr-4.8.2 。

PS2: 片段灵感来自 this answer 。

编辑: 我已经使用 pyspark 测试了以下内容。

16/11/22 07:16:56 INFO EventLoggingListener: Logging events to hdfs:///var/log/spark/apps/application_1479798580078_0001
16/11/22 07:16:56 INFO YarnClientSchedulerBackend: SchedulerBackend is ready for scheduling beginning after reached minRegisteredResourcesRatio: 0.8
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 1.6.2
      /_/

Using Python version 3.4.3 (default, Sep  1 2016 23:33:38)
SparkContext available as sc, HiveContext available as sqlContext.
>>> print(hash('foo'))
-2457967226571033580
>>> print(hash('foo'))
-2457967226571033580
>>> print(hash('foo'))
-2457967226571033580
>>> print(hash('foo'))
-2457967226571033580
>>> print(hash('foo'))
-2457967226571033580

还创建了一个简单的应用程序( simple_app.py ):

from pyspark import SparkContext

sc = SparkContext(appName = "simple-app")

numbers = [hash('foo') for i in range(10)]

print(numbers)

这似乎也很完美:

[hadoop@ip-*** ~]$ spark-submit --master yarn simple_app.py 

输出(截断):

[...]
16/11/22 07:28:42 INFO YarnClientSchedulerBackend: SchedulerBackend is ready for scheduling beginning after reached minRegisteredResourcesRatio: 0.8
[-5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594] // THE RELEVANT LINE IS HERE.
16/11/22 07:28:42 INFO SparkContext: Invoking stop() from shutdown hook
[...]

正如您所看到的，它也可以每次返回相同的哈希值。

编辑 2: 从评论来看，您似乎正在尝试计算执行程序而不是驱动程序的哈希值，因此您需要在 spark 应用程序配置中设置 spark.executorEnv.PYTHONHASHSEED ，以便它可以在执行程序上传播(这是一种方法)。

Note : Setting the environment variables for executors is the same with YARN client, use the spark.executorEnv.[EnvironmentVariableName].

因此，以下带有 simple_app.py 的极简示例:

from pyspark import SparkContext, SparkConf

conf = SparkConf().set("spark.executorEnv.PYTHONHASHSEED","123")
sc = SparkContext(appName="simple-app", conf=conf)

numbers = sc.parallelize(['foo']*10).map(lambda x: hash(x)).collect()

print(numbers)

现在让我们再次测试一下。这是截断的输出:

16/11/22 14:14:34 INFO DAGScheduler: Job 0 finished: collect at /home/hadoop/simple_app.py:6, took 14.251514 s
[-5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594, -5869373620241885594]
16/11/22 14:14:34 INFO SparkContext: Invoking stop() from shutdown hook

我认为这涵盖了所有。