Ich möchte eine CSV in Spark lesen und als DataFrame konvertieren und in HDFS mit speichern df.registerTempTable("table_name") Ich habe versucht: scala> val df = sqlContext.load("hdfs:///csv/file/dir/file.csv") Fehler, den ich bekommen habe: java.lang.RuntimeException:
Wie kann ich eine RDD konvertieren ( org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]) zu einem Datenrahmen org.apache.spark.sql.DataFrame. Ich habe einen Datenrahmen mit rdd konvertiert .rdd. Nach der Verarbeitung möchte ich es wieder im Datenrahmen haben. Wie kann ich das machen...
Ich möchte eine Spalte in a DataFramemit einem beliebigen Wert hinzufügen (das ist für jede Zeile gleich). Ich erhalte eine Fehlermeldung, wenn ich withColumnFolgendes verwende: dt.withColumn('new_column', 10).head(5) ---------------------------------------------------------------------------...
Ich habe es versucht, df.orderBy("col1").show(10)aber es wurde in aufsteigender Reihenfolge sortiert. df.sort("col1").show(10)sortiert auch in absteigender Reihenfolge. Ich habe mir den Stackoverflow angesehen und die Antworten, die ich gefunden habe, waren alle veraltet oder bezogen sich auf RDDs...
Was ist der Unterschied zwischen einem RDD map und einer mapPartitionsMethode? Und verhält flatMapsich wie mapoder wie mapPartitions? Vielen Dank. (bearbeiten) dh was ist der Unterschied (entweder semantisch oder in Bezug auf die Ausführung) zwischen def map[A, B](rdd: RDD[A], fn: (A => B))...
Ich habe begonnen, Spark SQL und DataFrames in Spark 1.4.0 zu verwenden. Ich möchte einen benutzerdefinierten Partitionierer in DataFrames in Scala definieren, sehe aber nicht, wie das geht. Eine der Datentabellen, mit denen ich arbeite, enthält eine Liste von Transaktionen nach Konto, silimar zum...
Ich habe einen Spark DataFrame (mit PySpark 1.5.1) und möchte eine neue Spalte hinzufügen. Ich habe Folgendes ohne Erfolg versucht: type(randomed_hours) # => list # Create in Python and transform to RDD new_col = pd.DataFrame(randomed_hours, columns=['new_col']) spark_new_col =...
Wie kann ich den für Apache Spark Executor-Knoten verfügbaren Speicher erhöhen? Ich habe eine 2-GB-Datei, die zum Laden in Apache Spark geeignet ist. Ich führe momentan Apache Spark auf einem Computer aus, sodass sich Treiber und Executor auf demselben Computer befinden. Das Gerät verfügt über 8 GB...
Ich versuche, den Inhalt einer Sammlung auf der Spark-Konsole zu drucken. Ich habe einen Typ: linesWithSessionId: org.apache.spark.rdd.RDD[String] = FilteredRDD[3] Und ich benutze den Befehl: scala> linesWithSessionId.map(line => println(line)) Aber das ist gedruckt: res1:
Wie verketten wir zwei Spalten in einem Apache Spark DataFrame? Gibt es eine Funktion in Spark SQL, die wir verwenden
Dies ist eine Kopie der Frage eines anderen in einem anderen Forum, die nie beantwortet wurde. Daher dachte ich, ich würde sie hier erneut stellen, da ich das gleiche Problem habe. (Siehe http://geekple.com/blogs/feeds/Xgzu7/posts/351703064084736 ) Ich habe Spark ordnungsgemäß auf meinem Computer...
Ich bin neu in Spark und versuche, CSV-Daten aus einer Datei mit Spark zu lesen. Folgendes mache ich: sc.textFile('file.csv') .map(lambda line: (line.split(',')[0], line.split(',')[1])) .collect() Ich würde erwarten, dass dieser Aufruf mir eine Liste der beiden ersten Spalten meiner Datei gibt,...
Ich verwende https://github.com/databricks/spark-csv . Ich versuche, eine einzelne CSV zu schreiben, kann dies jedoch nicht. Es wird ein Ordner erstellt. Benötigen Sie eine Scala-Funktion, die Parameter wie Pfad und Dateiname verwendet und diese CSV-Datei
Ich habe eine Spark-Streaming-Anwendung, die für jede Minute einen Datensatz erstellt. Ich muss die Ergebnisse der verarbeiteten Daten speichern / überschreiben. Beim Versuch, das Dataset org.apache.hadoop.mapred.FileAlreadyExistsException zu überschreiben, wird die Ausführung gestoppt. Ich habe...
Ich arbeite an einem Datenrahmen mit zwei Spalten, mvv und count. +---+-----+ |mvv|count| +---+-----+ | 1 | 5 | | 2 | 9 | | 3 | 3 | | 4 | 1 | Ich möchte zwei Listen mit MVV-Werten und Zählwert erhalten. Etwas wie mvv = [1,2,3,4] count = [5,9,3,1] Also habe ich den folgenden Code ausprobiert: Die...
Ich habe eine laufende Spark-Anwendung, in der alle Kerne belegt sind, in denen meinen anderen Anwendungen keine Ressource zugewiesen wird. Ich habe einige schnelle Nachforschungen angestellt und die Leute schlugen vor, YARN kill oder / bin / spark-class zu verwenden, um den Befehl zu beenden....
Ich folge dem großartigen Funken-Tutorial Also versuche ich um 46:00 Uhr, das zu laden, README.mdaber ich scheitere an dem, was ich tue, folgendes: $ sudo docker run -i -t -h sandbox sequenceiq/spark:1.1.0 /etc/bootstrap.sh -bash bash-4.1# cd /usr/local/spark-1.1.0-bin-hadoop2.4 bash-4.1# ls...
Im Moment muss ich df.count > 0überprüfen, ob das DataFrameleer ist oder nicht. Aber es ist irgendwie ineffizient. Gibt es einen besseren Weg, das zu tun? Vielen Dank. PS: Ich möchte überprüfen, ob es leer ist, damit ich das nur speichere, DataFramewenn es nicht leer
Ich habe einen Datenrahmen mit einer Spalte als String. Ich wollte den Spaltentyp in PySpark in Double-Typ ändern. Folgendes habe ich getan: toDoublefunc = UserDefinedFunction(lambda x: x,DoubleType()) changedTypedf = joindf.withColumn("label",toDoublefunc(joindf['show'])) Ich wollte nur wissen, ob...
Ich versuche, einen PySpark-Datenrahmen mit einem Zeilenwert zu filtern None: df.select('dt_mvmt').distinct().collect() [Row(dt_mvmt=u'2016-03-27'), Row(dt_mvmt=u'2016-03-28'), Row(dt_mvmt=u'2016-03-29'), Row(dt_mvmt=None), Row(dt_mvmt=u'2016-03-30'), Row(dt_mvmt=u'2016-03-31')] und ich kann mit...