Use df.select(cols.head, cols.tail: _*) Let me know if it works 🙂 Explanation from @Ben: The key is the method signature of select: select(col: String, cols: String*) The cols:String* entry takes a variable number of arguments. :_* unpacks arguments so that they can be handled by this argument. Very similar to unpacking in python with *args. … Read more
It seems like the problem is the amount of data you are trying to pull back to to your driver is too large. Most likely you are using the collect method to retrieve all values from a DataFrame/RDD. The driver is a single process and by collecting a DataFrame you are pulling all of that … Read more
count() can be used inside agg() as groupBy expression is same. With Python import pyspark.sql.functions as func new_log_df.cache().withColumn(“timePeriod”, encodeUDF(new_log_df[“START_TIME”])) .groupBy(“timePeriod”) .agg( func.mean(“DOWNSTREAM_SIZE”).alias(“Mean”), func.stddev(“DOWNSTREAM_SIZE”).alias(“Stddev”), func.count(func.lit(1)).alias(“Num Of Records”) ) .show(20, False) pySpark SQL functions doc With Scala import org.apache.spark.sql.functions._ //for count() new_log_df.cache().withColumn(“timePeriod”, encodeUDF(col(“START_TIME”))) .groupBy(“timePeriod”) .agg( mean(“DOWNSTREAM_SIZE”).alias(“Mean”), stddev(“DOWNSTREAM_SIZE”).alias(“Stddev”), count(lit(1)).alias(“Num Of Records”) ) .show(20, false) count(1) will count the … Read more
[*] Here is a simple illustrative experiment: import org.apache.spark.sql._ object SparkSandbox extends App { implicit val spark = SparkSession.builder().master(“local[*]”).getOrCreate() import spark.implicits._ spark.sparkContext.setLogLevel(“ERROR”) val left = Seq((1, “A1”), (2, “A2”), (3, “A3”), (4, “A4”)).toDF(“id”, “value”) val right = Seq((3, “A3”), (4, “A4”), (4, “A4_1”), (5, “A5”), (6, “A6”)).toDF(“id”, “value”) println(“LEFT”) left.orderBy(“id”).show() println(“RIGHT”) right.orderBy(“id”).show() val joinTypes = … Read more
Spark 2.2+ You can use explode_outer function: import org.apache.spark.sql.functions.explode_outer df.withColumn(“likes”, explode_outer($”likes”)).show // +—+—-+——–+ // | id|name| likes| // +—+—-+——–+ // | 1|Luke|baseball| // | 1|Luke| soccer| // | 2|Lucy| null| // +—+—-+——–+ Spark <= 2.1 In Scala but Java equivalent should be almost identical (to import individual functions use import static). import org.apache.spark.sql.functions.{array, col, explode, … Read more
It is not possible to create multiple top level columns from a single UDF call but you can create a new struct. It requires an UDF with specified returnType: from pyspark.sql.functions import udf from pyspark.sql.types import StructType, StructField, FloatType schema = StructType([ StructField(“foo”, FloatType(), False), StructField(“bar”, FloatType(), False) ]) def udf_test(n): return (n / 2, … Read more
Use countDistinct function from pyspark.sql.functions import countDistinct x = [(“2001″,”id1”),(“2002″,”id1”),(“2002″,”id1”),(“2001″,”id1”),(“2001″,”id2”),(“2001″,”id2”),(“2002″,”id2”)] y = spark.createDataFrame(x,[“year”,”id”]) gr = y.groupBy(“year”).agg(countDistinct(“id”)) gr.show() output +—-+——————+ |year|count(DISTINCT id)| +—-+——————+ |2002| 2| |2001| 2| +—-+——————+
The short answer is, there’s no “accepted” way to do this, but you can do it very elegantly with a recursive function that generates your select(…) statement by walking through the DataFrame.schema. The recursive function should return an Array[Column]. Every time the function hits a StructType, it would call itself and append the returned Array[Column] … Read more
Scala (see below for PySpark) The spark-fast-tests library has two methods for making DataFrame comparisons (I’m the creator of the library): The assertSmallDataFrameEquality method collects DataFrames on the driver node and makes the comparison def assertSmallDataFrameEquality(actualDF: DataFrame, expectedDF: DataFrame): Unit = { if (!actualDF.schema.equals(expectedDF.schema)) { throw new DataFrameSchemaMismatch(schemaMismatchMessage(actualDF, expectedDF)) } if (!actualDF.collect().sameElements(expectedDF.collect())) { throw new … Read more
I noticed that your problematic line has escaping that uses double quotes themselves: “32 XIY “”W”” JK, RE LK” which should be interpreter just as 32 XIY “W” JK, RE LK As described in RFC-4180, page 2 – If double-quotes are used to enclose fields, then a double-quote appearing inside a field must be escaped … Read more