As of Spark 1.4, you can use Transformer org.apache.spark.ml.feature.VectorAssembler. Just provide column names you want to be features. val assembler = new VectorAssembler() .setInputCols(Array(“col1”, “col2”, “col3”)) .setOutputCol(“features”) and add it to your pipeline.
This is where Optioncomes in handy: val extractDateAsOptionInt = udf((d: String) => d match { case null => None case s => Some(s.substring(0, 10).filterNot(“-“.toSet).toInt) }) or to make it slightly more secure in general case: import scala.util.Try val extractDateAsOptionInt = udf((d: String) => Try( d.substring(0, 10).filterNot(“-“.toSet).toInt ).toOption) All credit goes to Dmitriy Selivanov who’ve pointed … Read more
It depends on the type of your “list”: If it is of type ArrayType(): df = hc.createDataFrame(sc.parallelize([[‘a’, [1,2,3]], [‘b’, [2,3,4]]]), [“key”, “value”]) df.printSchema() df.show() root |– key: string (nullable = true) |– value: array (nullable = true) | |– element: long (containsNull = true) you can access the values like you would with python using … Read more
groupBy after orderBy doesn’t maintain order, as others have pointed out. What you want to do is use a Window function, partitioned on id and ordered by hours. You can collect_list over this and then take the max (largest) of the resulting lists since they go cumulatively (i.e. the first hour will only have itself … Read more
I think you are looking for RDD.union val rddPart1 = ??? val rddPart2 = ??? val rddAll = rddPart1.union(rddPart2) Example (on Spark-shell) val rdd1 = sc.parallelize(Seq((1, “Aug”, 30),(1, “Sep”, 31),(2, “Aug”, 15),(2, “Sep”, 10))) val rdd2 = sc.parallelize(Seq((1, “Oct”, 10),(1, “Nov”, 12),(2, “Oct”, 5),(2, “Nov”, 15))) rdd1.union(rdd2).collect res0: Array[(Int, String, Int)] = Array((1,Aug,30), (1,Sep,31), (2,Aug,15), … Read more
How to create a new column in PySpark and fill this column with the date of today? There is already function for that: from pyspark.sql.functions import current_date df.withColumn(“date”, current_date().cast(“string”)) AssertionError: col should be Column Use literal from pyspark.sql.functions import lit df.withColumn(“date”, lit(str(now)[:10]))
If you want a single file, you need to do a coalesce to a single partition before calling write, so: unzipJSON.coalesce(1).write.mode(“append”).json(“/home/eranw/Workspace/JSON/output/unCompressedJson.json”) Personally, I find it rather annoying that the number of output files depend on number of partitions you have before calling write – especially if you do a write with a partitionBy – but … Read more
This uses last and ignores nulls. Let’s re-create something similar to the original data: import sys from pyspark.sql.window import Window import pyspark.sql.functions as func d = [{‘session’: 1, ‘ts’: 1}, {‘session’: 1, ‘ts’: 2, ‘id’: 109}, {‘session’: 1, ‘ts’: 3}, {‘session’: 1, ‘ts’: 4, ‘id’: 110}, {‘session’: 1, ‘ts’: 5}, {‘session’: 1, ‘ts’: 6}] df … Read more
The Spark union is implemented according to standard SQL and therefore resolves the columns by position. This is also stated by the API documentation: Return a new DataFrame containing union of rows in this and another frame. This is equivalent to UNION ALL in SQL. To do a SQL-style set union (that does >deduplication of … Read more