apache-spark-sql

How to calculate date difference in pyspark?

by

You need to cast the column low to class date and then you can use datediff() in combination with lit(). Using Spark 2.2: from pyspark.sql.functions import datediff, to_date, lit df.withColumn(“test”, datediff(to_date(lit(“2017-05-02”)), to_date(“low”,”yyyy/MM/dd”))).show() +———-+—-+——+—–+ | low|high|normal| test| +———-+—-+——+—–+ |1986/10/15| z| null|11157| |1986/10/15| z| null|11157| |1986/10/15| c| null|11157| |1986/10/15|null| null|11157| |1986/10/16|null| 4.0|11156| +———-+—-+——+—–+ Using < Spark 2.2, … Read more

How to express a column which name contains spaces in Spark SQL?

by

Backticks seem to work just fine: scala> val df = sc.parallelize(Seq((“a”, 1))).toDF(“foo bar”, “x”) df: org.apache.spark.sql.DataFrame = [foo bar: string, x: int] scala> df.registerTempTable(“df”) scala> sqlContext.sql(“””SELECT `foo bar` FROM df”””).show foo bar a Same as DataFrame API: scala> df.select($”foo bar”).show foo bar a So it looks like it is supported, although I doubt it is … Read more

Does spark.sql.autoBroadcastJoinThreshold work for joins using Dataset’s join operator?

by

First of all spark.sql.autoBroadcastJoinThreshold and broadcast hint are separate mechanisms. Even if autoBroadcastJoinThreshold is disabled setting broadcast hint will take precedence. With default settings: spark.conf.get(“spark.sql.autoBroadcastJoinThreshold”) String = 10485760 val df1 = spark.range(100) val df2 = spark.range(100) Spark will use autoBroadcastJoinThreshold and automatically broadcast data: df1.join(df2, Seq(“id”)).explain == Physical Plan == *Project [id#0L] +- *BroadcastHashJoin [id#0L], … Read more

Avoid performance impact of a single partition mode in Spark window functions

by

In practice performance impact will be almost the same as if you omitted partitionBy clause at all. All records will be shuffled to a single partition, sorted locally and iterated sequentially one by one. The difference is only in the number of partitions created in total. Let’s illustrate that with an example using simple dataset … Read more

SPARK SQL – update MySql table using DataFrames and JDBC

by

It is not possible. As for now (Spark 1.6.0 / 2.2.0 SNAPSHOT) Spark DataFrameWriter supports only four writing modes: SaveMode.Overwrite: overwrite the existing data. SaveMode.Append: append the data. SaveMode.Ignore: ignore the operation (i.e. no-op). SaveMode.ErrorIfExists: default option, throw an exception at runtime. You can insert manually for example using mapPartitions (since you want an UPSERT … Read more

Explode (transpose?) multiple columns in Spark SQL table

by

Spark >= 2.4 You can skip zip udf and use arrays_zip function: df.withColumn(“vars”, explode(arrays_zip($”varA”, $”varB”))).select( $”userId”, $”someString”, $”vars.varA”, $”vars.varB”).show Spark < 2.4 What you want is not possible without a custom UDF. In Scala you could do something like this: val data = sc.parallelize(Seq( “””{“userId”: 1, “someString”: “example1”, “varA”: [0, 2, 5], “varB”: [1, 2, … Read more

Queries with streaming sources must be executed with writeStream.start();

by

You are branching the query plan: from the same ds1 you are trying to: ds1.collect.foreach(…) ds1.writeStream.format(…){…} But you are only calling .start() on the second branch, leaving the other dangling without a termination, which in turn throws the exception you are getting back. The solution is to start both branches and await termination. val ds1 … Read more

Passing Array to Spark Lit function

by

List comprehension inside Spark’s array a = [1,2,3,4,5,6,7,8,9,10] df = spark.createDataFrame([[‘a b c d e f g h i j ‘],], [‘col1’]) df = df.withColumn(“NewColumn”, F.array([F.lit(x) for x in a])) df.show(truncate=False) df.printSchema() # +——————–+——————————-+ # |col1 |NewColumn | # +——————–+——————————-+ # |a b c d e f g h i j |[1, 2, 3, 4, … Read more