Sampling Data with SparkSql Examples using explode functions in Map and Array

-

What is explode function

Spark SQL explode function is used to create or split an array or map DataFrame columns to rows. Spark defines several flavors of this function; explode_outer – to handle nulls and empty, posexplode – which explodes with a position of element and posexplode_outer – to handle nulls.

Difference between explode vs explode_outer

explode – creates a row for each element in the array or map column by ignoring null or empty values in array. whereas explode_outer returns all values in array or map including null or empty.

Difference between explode vs posexplode

explode – creates a row for each element in the array or map column. whereas posexplode creates a row for each element in the array and creates two columns ‘pos’ to hold the position of the array element and the ‘col’ to hold the actual array value. And, for the map, it creates 3 columns ‘pos’, ‘key’ and ‘value’

package com.dpq.spark.dataframe.functionsimport com.dpq.spark.dataframe.functions.ArraTypeExample.sparkimport org.apache.spark.sql.{Row, SparkSession}import org.apache.spark.sql.functions._import org.apache.spark.sql.types.{ArrayType, IntegerType, MapType, StringType, StructType}object ExplodeArrayAndMap{  def main(args:Array[String]) : Unit = {    val spark = SparkSession.builder().appName("dpq.spark.example.com")      .master("local[1]")      .getOrCreate()     // create DataFrame    val arrayData = Seq(      Row("deepak",List("Java","Scala","C++"),Map("hair"->"black","eye"->"brown")),      Row("anant",List("Spark","Java","C++",null),Map("hair"->"brown","eye"->null)),      Row("lokesh",List("CSharp","Python",""),Map("hair"->"red","eye"->"")),      Row("pune",null,null),      Row("Jeferson",List(),Map())    )    val arraySchema = new StructType()      .add("name",StringType)      .add("knownLanguages", ArrayType(StringType))      .add("properties", MapType(StringType,StringType))    val df = spark.createDataFrame(spark.sparkContext.parallelize(arrayData),arraySchema)    df.printSchema()    df.show()    import spark.implicits._    // Below are Array examples    //explode    df.select($"name",explode($"knownLanguages"))      .show()    //explode_outer    df.select($"name",explode_outer($"knownLanguages"))      .show()    //posexplode    df.select($"name",posexplode($"knownLanguages"))      .show()    //posexplode_outer    df.select($"name",posexplode_outer($"knownLanguages"))      .show()    // Below are Map examples    //explode    df.select($"name",explode($"properties"))      .show()    //explode_outer    df.select($"name",explode_outer($"properties"))      .show()    //posexplode    df.select($"name",posexplode($"properties"))      .show()    //posexplode_outer    df.select($"name",posexplode_outer($"properties"))      .show()  }}

Popular posts from this blog

How to change column name in Dataframe and selection of few columns in Dataframe using Pyspark with example

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more

What is Garbage collection in Spark and its impact and resolution

-
Image
Garbage Collection Spark runs on the Java Virtual Machine (JVM). Because Spark can store large amounts of data in memory, it has a major reliance on Java’s memory management and garbage collection (GC). Therefore, garbage collection  (GC) can be a major issue that can affect many Spark applications. Common symptoms of excessive GC in Spark are: Slowness of application Executor heartbeat timeout GC overhead limit exceeded error Spark’s memory-centric approach and data-intensive applications make it a more common issue than other Java applications. Thankfully, it’s easy to diagnose if your Spark application is suffering from a GC problem. The Spark UI marks executors in red if they have spent too much time doing GC. Spark executors are spending a significant amount of CPU cycles performing garbage collection. This can be determined by looking at the “Executors” tab in the Spark application UI. Spark will mark an executor in  red  if the executor has spent more than 10% of the time in gar...
Read more

Window function in PySpark with Joins example using 2 Dataframes (inner join)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more