AmazonのクラスタでPySparkが非常に遅い

Question

私はSparkプロジェクトに取り組んでおり、アマゾンのクラスタでアプリをエスケープしようとしています。小さなファイルでもパフォーマンスは非常に遅いです。私は解決策を望んでいない、経験豊富な遅さのための考えられる理由についてのちょうど意見。

spark = SparkSession.builder.appName("appName").getOrCreate() 
sc = spark.sparkContext 

rec= sc.textFile(sys.argv[1]) 
# rec= sc.parallelize(records.collect()) 

a= rec.map(lambda line: line.split("\t")) 
      .filter(lambda x: int(x[6])>=4) 
      .map(lambda x: (x[1],[x[2], x[6]])) 
a=a.join(a) 
    .filter(lambda (x,(a,b)): a[0]<b[0]) 
    .map(lambda (x,(b,a)): ((a[0],b[0]),x)) 
    .groupByKey() 
    .filter(lambda (x,y): len(set(y))>2) 
    .sortBy(lambda a: a[0]) 
    .saveAsTextFile(sys.argv[2]) 

Answer 1

このコードシーケンスは、あなたの最も深い悲しみを引き起こしている可能性があります

records = sc.textFile(sys.argv[1]) 
rec= sc.parallelize(records.collect()) 

なぜ？あなたはスパークコンテキストのテキストファイル関数を介してRDDとしてファイルを読み込んでいます。そして、それはクラスタ全体にマテリアライズされます。次に、キッカーがあります。records.collect（）を呼び出すと、クラスタにすべてのデータを送信するように指示します。 （どのマシンがジョブを起動したとしても）ドライバに渡し、最後に、ローカルに収集されたリストからRDDを再構築します。 rec RDDの代わりにRDDレコードを使用してください

編集： aを自分自身にクロス結合しているようです。それは意図されているのですか？

キーでグループ化すると、すべてのデータを強制的にシャッフルし、キーで縮小します。

SparkContextの使用を中止します。下位互換性のためにそこにあります。あなたのsparksessionの.read.option("delimiter", "\t").csv(file path)を使って、RDDの代わりにDataFrameを作成し、タブ区切りの行を一般的なRowオブジェクトに解析し、DataFrame APIを使用して非常に優れたパフォーマンスを得ることができます（タングステンとCatalystによる）。これはPySparkを使用しているので、Spark 2.xをDataFramesとPythonで使用すると、パフォーマンスがJVMのスカラと一致するため、PythonでRDDを使用すると、Pythonインタプリタが作業を行うため、Scalaの方がはるかに高速です）。

Answer 2

よくPySparkまたはSparkは、大量のデータを処理する必要がある場合に使用します。小さなデータでは遅いことが予想されます。理由は次のとおりです。

1. 実行中のエグゼキュータの数に応じてPySparkでは、JVMを最初に起動する必要があります。さらに、PySparkの場合、呼び出される必要があるPythonのサブプロセスを通じて追加のオーバーヘッドが作成されます。
2. 第2の理由は、データシャッフルによるものです。あなたのデータはネットワーク上でシャッフルされるかもしれません。ローカルの場合、データは同じノード上で計算されます。データの配布のために、スケジューラはまずデータを配置する必要がある場所を特定し、次にそれをどのように処理するかを把握する必要があります。

PySpark/Sparkは、「大きなデータ」で何かをする必要がある場合にのみ輝きます！スパークは遅いと言われているので、最初は非常に多くの人が非常に失望しているのを見ましたが、ごくわずかなデータ量でしか使用していませんでした。お役に立てれば！一人で