These are not intended to work in the same way. groupBy is simply an equivalent of the GROUP BY clause in standard SQL. In other words table.groupBy($”foo”, $”bar”) is equivalent to: SELECT foo, bar, [agg-expressions] FROM table GROUP BY foo, bar cube is equivalent to CUBE extension to GROUP BY. It takes a list of … Read more
It is simple: partitionColumn is a column which should be used to determine partitions. lowerBound and upperBound determine range of values to be fetched. Complete dataset will use rows corresponding to the following query: SELECT * FROM table WHERE partitionColumn BETWEEN lowerBound AND upperBound numPartitions determines number of partitions to be created. Range between lowerBound … Read more
Indexing input data The fundamental reason why indexing over external data sources is not in the Spark scope is that Spark is not a data management system but a batch data processing engine. Since it doesn’t own the data it is using it cannot reliably monitor changes and as a consequence cannot maintain indices. If … Read more
The source of the problem is that object returned from the UDF doesn’t conform to the declared type. np.unique not only returns numpy.ndarray but also converts numerics to the corresponding NumPy types which are not compatible with DataFrame API. You can try something like this: udf(lambda x: list(set(x)), ArrayType(IntegerType())) or this (to keep order) udf(lambda … Read more
You can do this without a udf using a Window. Consider the following example: import pyspark.sql.functions as f data = [ (‘a’, 5), (‘a’, 8), (‘a’, 7), (‘b’, 1), (‘b’, 3) ] df = sqlCtx.createDataFrame(data, [“A”, “B”]) df.show() #+—+—+ #| A| B| #+—+—+ #| a| 5| #| a| 8| #| a| 7| #| b| 1| … Read more
I had the exact same problem and I found a way to do this using DataFrame.repartition(). The problem with using coalesce(1) is that your parallelism drops to 1, and it can be slow at best and error out at worst. Increasing that number doesn’t help either — if you do coalesce(10) you get more parallelism, … Read more
This can be done using a combination of a window function and the Window.unboundedPreceding value in the window’s range as follows: from pyspark.sql import Window from pyspark.sql import functions as F windowval = (Window.partitionBy(‘class’).orderBy(‘time’) .rangeBetween(Window.unboundedPreceding, 0)) df_w_cumsum = df.withColumn(‘cum_sum’, F.sum(‘value’).over(windowval)) df_w_cumsum.show() +—-+—–+—–+——-+ |time|value|class|cum_sum| +—-+—–+—–+——-+ | 1| 3| b| 3| | 2| 3| b| 6| | … Read more
Import lower alongside col: from pyspark.sql.functions import lower, col Combine them together using lower(col(“bla”)). In a complete query: spark.table(‘bla’).select(lower(col(‘bla’)).alias(‘bla’)) which is equivalent to the SQL query SELECT lower(bla) AS bla FROM bla To keep the other columns, do spark.table(‘foo’).withColumn(‘bar’, lower(col(‘bar’))) Needless to say, this approach is better than using a UDF because UDFs have to … Read more
Actions vs Transformations Collect (Action) – Return all the elements of the dataset as an array at the driver program. This is usually useful after a filter or other operation that returns a sufficiently small subset of the data. spark-sql doc select(*cols) (transformation) – Projects a set of expressions and returns a new DataFrame. Parameters: … Read more
toDF method is a monkey patch executed inside SparkSession (SQLContext constructor in 1.x) constructor so to be able to use it you have to create a SQLContext (or SparkSession) first: # SQLContext or HiveContext in Spark 1.x from pyspark.sql import SparkSession from pyspark import SparkContext sc = SparkContext() rdd = sc.parallelize([(“a”, 1)]) hasattr(rdd, “toDF”) ## … Read more