AWS EMRでPYTHONHASHSEEDを設定する方法

Question

EMRクラスタのすべてのノードで環境変数を設定する方法はありますか？

Python3 PySparkでreduceByKey（）を使用しようとしたときにエラーが発生し、ハッシュシードに関するエラーが発生しました。私はこれが既知のエラーであることを知ることができ、環境のvarialbe PYTHONHASHSEEDはクラスタのすべてのノードで同じ値に設定する必要がありますが、私はそれに幸運はありませんでした。

私は、クラスタ構成による-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あなたがspark-envスコープの下でクラスタ構成で定義されている環境変数PYTHONHASHSEEDを拾っていないと信じています。

あなたが代わりに/usr/bin/python3のpython34を使用し、以下のように設定を設定はず：

[ 
    { 
     "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: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 
[...] 

：

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: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）を作成しましたあなたはできる毎回同じハッシュを返すこともできます。

EDIT 2：コメントから、あなたが執行していないドライバーのハッシュを計算しようとしているようにそれが伝播することができますので、このようにあなたはスパークアプリケーション構成内で、spark.executorEnv.PYTHONHASHSEEDを設定する必要がありますようですエグゼクティブに（それを行う方法の1つです）。

注：執行者のための環境変数を設定すると、このようにsimple_app.pyと、次のミニマリストの例をspark.executorEnv.[EnvironmentVariableName].

を使用し、糸クライアントと同じです：

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 

これはすべてをカバーすると思います。

Answer 2

おそらく、ブートストラップスクリプト経由でそれを行うことができますが、このような何かする必要があります：それは時にスパークプロセスによってピックアップされますようにするため

echo "PYTHONHASHSEED=XXXX" >> /home/hadoop/.bashrc 

（または多分.profile）

を彼らは立ち上げられる。

あなたの設定は合理的ですが、代わりにhadoop-envセクションに設定する価値がありますか？スパークから

Answer 3

docs

注：クラスタモードでYARN上でスパークを実行する場合、環境変数はspark.yarn.appMasterEnvあなたのconf /火花デフォルトで[EnvironmentVariableName]プロパティを使用して設定する必要があります.confファイル。 spark-env.shで設定された環境変数は、クラスタモードのYARNアプリケーションマスタープロセスに反映されません。詳細については、YARN関連のSpark Propertiesを参照してください。

プロパティはhereを列挙されているので、私はあなたがこのほしいと思う：

がYARNに起動されたアプリケーションのマスター・プロセスにEnvironmentVariableNameで指定した環境変数を追加します。スパークdefaults.confにを設定するため

spark.yarn.appMasterEnv.PYTHONHASHSEED="XXXX" 

EMRのドキュメントはhereあります。

[ 
    { 
     "Classification": "spark-defaults", 
     "Properties": { 
     "spark.yarn.appMasterEnv.PYTHONHASHSEED: "XXX" 
     } 
    } 
] 

Answer 4

ただ、次のような構成は、それを解決し追加して、同じ問題が発生しました：

# Some settings... 
Configurations=[ 
     { 
      "Classification": "spark-env", 
      "Properties": {}, 
      "Configurations": [ 
       { 
        "Classification": "export", 
        "Properties": { 
         "PYSPARK_PYTHON": "python34" 
        }, 
        "Configurations": [] 
       } 
      ] 
     }, 
     { 
      "Classification": "hadoop-env", 
      "Properties": {}, 
      "Configurations": [ 
       { 
        "Classification": "export", 
        "Properties": { 
         "PYTHONHASHSEED": "0" 
        }, 
        "Configurations": [] 
       } 
      ] 
     } 
     ], 
# Some more settings... 

は注意してください：我々は、クラスタが唯一のHadoopとスパークを実行している瞬間のために、クラスタマネージャとしての糸を使用しないでください。

EDIT：Tim Bのコメントに続いて、これはクラスタマネージャーとしてインストールされた糸でも機能するようです。