Wie ich in Spark Dataframe weiß, kann dieser für mehrere Spalten denselben Namen haben, wie im folgenden Datenrahmen-Snapshot gezeigt: [ Row(a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0}), a=107831, f=SparseVector(5, {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0})), Row(a=107831,...
Ich habe versucht, einen vernünftigen Weg zum Testen SparkSessionmit dem JUnit-Testframework zu finden. Obwohl es gute Beispiele dafür zu geben scheint SparkContext, konnte ich nicht herausfinden, wie ein entsprechendes Beispiel zum Laufen gebracht werden kann SparkSession, obwohl es an mehreren...
Ich verwende Spark 1.3.1 (PySpark) und habe mithilfe einer SQL-Abfrage eine Tabelle generiert. Ich habe jetzt ein Objekt, das a ist DataFrame. Ich möchte dieses DataFrameObjekt (ich habe es "Tabelle" genannt) in eine CSV-Datei exportieren, damit ich es bearbeiten und die Spalten zeichnen kann....
Gibt es eine Möglichkeit, eine Aggregatfunktion auf alle (oder eine Liste von) Spalten eines Datenrahmens anzuwenden, wenn Sie a ausführen groupBy? Mit anderen Worten, gibt es eine Möglichkeit, dies für jede Spalte zu vermeiden: df.groupBy("col1") .agg(sum("col2").alias("col2"),...
Gibt es eine Möglichkeit, die aktuelle Anzahl der Partitionen eines DataFrame abzurufen? Ich habe das DataFrame-Javadoc (Spark 1.6) überprüft und keine Methode dafür gefunden, oder habe ich es einfach verpasst? (Im Fall von JavaRDD gibt es eine getNumPartitions () -Methode.)
Bei Betrachtung der neuen Spark-Datenrahmen-API ist unklar, ob es möglich ist, Datenrahmenspalten zu ändern. Wie würde ich mich über einen Wert in der Zeile zu ändern xSpalte yeines Datenrahmens? In pandasdiesem wäredf.ix[x,y] = new_value Bearbeiten: Wenn Sie das unten Gesagte konsolidieren,...
val items = List("a", "b", "c") sqlContext.sql("select c1 from table") .filter($"c1".isin(items)) .collect .foreach(println) Der obige Code löst die folgende Ausnahme aus. Exception in thread "main" java.lang.RuntimeException: Unsupported literal type class scala.collection.immutable.$colon$colon...
Ich versuche, meine RDD mit Off-Heap-Speicher auf Spark 1.4.0 und Tachyon 0.6.4 so zu halten: val a = sqlContext.parquetFile("a1.parquet") a.persist(org.apache.spark.storage.StorageLevel.OFF_HEAP) a.count() Danach bekomme ich folgende Ausnahme. Irgendwelche Ideen dazu? 15/06/16 10:14:53...
Das Ziel dieser Frage ist es zu dokumentieren: Schritte zum Lesen und Schreiben von Daten über JDBC-Verbindungen in PySpark mögliche Probleme mit JDBC-Quellen und bekannten Lösungen Mit kleinen Änderungen sollten diese Methoden mit anderen unterstützten Sprachen wie Scala und R...
Was wäre der effizienteste Weg, um Millionen von Datensätzen einzufügen, beispielsweise 50 Millionen von einem Spark-Datenrahmen in Postgres-Tabellen. Ich habe dies in der Vergangenheit von Spark bis MSSQL getan, indem ich die Option für Massenkopien und Stapelgrößen verwendet habe, die ebenfalls...
Ich habe einen Datenrahmen mit folgendem Code: def test(lat: Double, lon: Double) = { println(s"testing ${lat / lon}") Map("one" -> "one", "two" -> "two") } val testUDF = udf(test _) df.withColumn("test", testUDF(col("lat"), col("lon"))) .withColumn("test1", col("test.one"))...