Ich bevorzuge Python gegenüber Scala. Da Spark jedoch ursprünglich in Scala geschrieben wurde, hatte ich aus offensichtlichen Gründen damit gerechnet, dass mein Code in Scala schneller ausgeführt wird als in der Python-Version.

Mit dieser Annahme wollte ich die Scala-Version eines sehr gebräuchlichen Vorverarbeitungscodes für etwa 1 GB Daten lernen und schreiben. Die Daten stammen vom SpringLeaf-Wettbewerb auf Kaggle. Nur um einen Überblick über die Daten zu geben (es enthält 1936 Dimensionen und 145232 Zeilen). Daten bestehen aus verschiedenen Typen, z.B. int, float, string, boolean Ich verwende 6 von 8 Kernen für die Spark-Verarbeitung. deshalb habe ich @ verwendminPartitions=6 Damit jeder Kern etwas zu verarbeiten hat.

Scala Code

val input = sc.textFile("train.csv", minPartitions=6)

val input2 = input.mapPartitionsWithIndex { (idx, iter) => 
  if (idx == 0) iter.drop(1) else iter }
val delim1 = "\001"

def separateCols(line: String): Array[String] = {
  val line2 = line.replaceAll("true", "1")
  val line3 = line2.replaceAll("false", "0")
  val vals: Array[String] = line3.split(",")

  for((x,i) <- vals.view.zipWithIndex) {
    vals(i) = "VAR_%04d".format(i) + delim1 + x
  }
  vals
}

val input3 = input2.flatMap(separateCols)

def toKeyVal(line: String): (String, String) = {
  val vals = line.split(delim1)
  (vals(0), vals(1))
}

val input4 = input3.map(toKeyVal)

def valsConcat(val1: String, val2: String): String = {
  val1 + "," + val2
}

val input5 = input4.reduceByKey(valsConcat)

input5.saveAsTextFile("output")

Python Code

input = sc.textFile('train.csv', minPartitions=6)
DELIM_1 = '\001'


def drop_first_line(index, itr):
  if index == 0:
    return iter(list(itr)[1:])
  else:
    return itr

input2 = input.mapPartitionsWithIndex(drop_first_line)

def separate_cols(line):
  line = line.replace('true', '1').replace('false', '0')
  vals = line.split(',')
  vals2 = ['VAR_%04d%s%s' %(e, DELIM_1, val.strip('\"'))
           for e, val in enumerate(vals)]
  return vals2


input3 = input2.flatMap(separate_cols)

def to_key_val(kv):
  key, val = kv.split(DELIM_1)
  return (key, val)
input4 = input3.map(to_key_val)

def vals_concat(v1, v2):
  return v1 + ',' + v2

input5 = input4.reduceByKey(vals_concat)
input5.saveAsTextFile('output')

Scala Performance Stufe 0 (38 Minuten), Stufe 1 (18 Sekunden)

Python-Leistung Stufe 0 (11 Minuten), Stufe 1 (7 Sekunden)

Both erzeugt verschiedene DAG-Visualisierungsgraphen (beide Bilder zeigen unterschiedliche Stage 0-Funktionen für Scala map) und Python reduceByKey))

Aber im Wesentlichen versuchen beide Codes, Daten in (dimension_id, Zeichenfolge der Werteliste) RDD umzuwandeln und auf der Festplatte zu speichern. Die Ausgabe wird verwendet, um verschiedene Statistiken für jede Dimension zu berechnen.

Performance weise, scheint Scala-Code für diese realen Daten wie folgt zu laufen 4 mal langsamer als die Python-Version. Eine gute Nachricht für mich ist, dass es mir eine gute Motivation gab, bei Python zu bleiben. Eine schlechte Nachricht ist, dass ich nicht ganz verstanden habe, warum?