spark notes

Read all list of list in spark

import  org.apache.hadoop.fs.{FileSystem,Path}

FileSystem.get( sc.hadoopConfiguration ).listStatus( new Path("/shravan/json")).foreach( x => println(x.getPath )) 

#######

find Duilpcate and read HDFS files

val df = spark.read.option("header", "true").option("delimiter", "|").option("inferSchema", "true").csv("a.scv")

#find dulpicate

 df.groupBy("id","name","age").count().filter("count >1").coalesce(1).write.format("csv").mode("overwrite").save("/tmp/spark_output/datacsv1")

 

#########################3

Write  to hgfs

******use************

 **df2.coalesce(1).write.mode("overwrite").insertInto("db.tb")

df1.coalesce(1).write.format("orc").mode("overwrite").saveAsTable("db.tb");

df.coalesce(1).write.mode("overwrite").csv("/shravan/")

df.write.format("csv").save("/tmp/spark_output/datacsv")

df .coalesce(1) .write.mode("overwrite").option("header","true").csv("/shravan/"

######################

spark-shell --master yarn --conf spark.ui.port=0 --num-executors 50 --driver-memory 40g --executor-memory 20g --total-executor-cores 10 --queue xyz--conf spark.sql.crossJoin.enabled=true

df = df.repartition(1000)

spark.dynamicAllocation.enabled true

 spark.executor.memory 11168M

 spark.executor.cores 4

####################################################

Read properties

import java.util.Properties

import org.apache.hadoop.conf.Configuration

import org.apache.hadoop.fs.{FileSystem, Path}

  def readPropertiesFile(propFilePath: String): Properties = {

    val props = new Properties()

    try {

      val hdfsConf: Configuration = new Configuration();

      val fs = FileSystem.get(hdfsConf);

      val is = fs.open(new Path(propFilePath));

      props.load(is)

      if (props == null) throw new Exception("Properties object not created")

    } catch {

      case allException: Exception => {

        error(s"Error: Generic exception ${allException.printStackTrace}")

      }

    }

    props

  }

val propObject = readPropertiesFile("abc.properties")

hive with clause

the subquery twice or once) but it works. Here is the fiddle

WITH subquery as
       ( 
SELECT
    First.deal_id
    ,COALESCE( First.exchange, Second.exchange ) as exchange_id
    ,COALESCE( First.publisher, Second.publisher ) as pub_id
    ,COUNT (DISTINCT(case when Second.user_id is null then First.user_id else null END)) AS Incremental
    ,SUM (First.imps) AS First_imps
    ,SUM (Second.imps) AS Second_imps
    FROM
        (
            SELECT
                a.deal_id
                ,a.exchange
                ,a.publisher
                ,a.user_id
                ,1 AS imps
            FROM
                T1 a 
            WHERE
                a.deal_id >= 1
        ) First 
        FULL OUTER JOIN (
            SELECT
                a.exchange
                ,a.publisher
                ,a.user_id
                ,1 AS imps
            FROM
                T1 a
            WHERE
            a.deal_id = 0
        ) Second
        ON (
            First.exchange = Second.exchange
            AND First.publisher = Second.publisher
            AND First.user_id = Second.user_id
        )
        GROUP BY
        First.deal_id
        ,COALESCE( First.exchange, Second.exchange )
        ,COALESCE( First.publisher, Second.publisher )
        )

SELECT
deal.deal_id,
deal.exchange_id,
deal.pub_id,
sum(deal.incremental),
sum(deal.first_imps),
sum(coalesce(deal.second_imps, 0) + coalesce(oa.second_imps,0))
FROM 
subquery deal
LEFT JOIN 
subquery oa 
ON (deal.exchange_id = oa.exchange_id
AND deal.pub_id = oa.pub_id
AND oa.deal_id is null)
WHERE deal.deal_id is not null
GROUP BY
deal.deal_id,
deal.exchange_id,
deal.pub_id
;

hive nvl, colease, when, if

Hadoop Hive supports the various Conditional functions such as IF, CASE, COALESCE, NVL, DECODE etc. You can use these function for testing equality, comparison operators and check if value is null. Following diagram shows various Hive Conditional Functions:

Hive Conditional Functions

Below table describes the various Hive conditional functions:

Conditional Function	Description
IF(boolean testCondition, T valueTrue, T valueFalseOrNull);	This is the one of best Hive Conditional Functions and is similar to the IF statements in other programming languages. The IF Hive Conditional functions tests an expression and returns a corresponding result depending on whether the result is true, false or null.
isnull( a )	Returns true if a is NULL and false otherwise.
isnotnull ( a )	This function is a negation of isnull function. Returns true if a is not NULL and false otherwise.
CASE WHEN a THEN b [WHEN c THEN d]… [ELSE e] END	Function Tests whether any of a sequence of expressions is true, and returns a corresponding result for the first true expression. You can use OR, IN, REGEXP in the CASE expressions.
CASE a WHEN b THEN c [WHEN d THEN e]… [ELSE f] END	Function Tests whether any of a sequence of expressions is true, and returns a corresponding result for the first true expression. You can use OR, IN, REGEXP in the CASE expressions.
NVL(arg1, arg2)	This function will test if expression is null, it’ll return expression if result is not null otherwise second argument is returned.
coalesce(value1, value 2, …)	Returns the first non-null value for list of values provided as arguments.
decode(<expr>, <search1>,<result1>, …<search N>, <result N>, <default>)	Compares an expression to one or more possible values, and returns a corresponding result when a match is found.

Hive Conditional Functions Examples

Hive IF condition function example

select IF(1=1,'TRUE','FALSE') as IF_CONDITION_TEST;

Hive ISNULL condition function example

select isnull( NULL );

Hive ISNOTNULL condition function example

select isnotnull( NULL );

Hive CASE conditional function examples

select case x 
when 1 then 'one' 
when 2 then 'two' 
when 0 then 'zero' 
else 'out of range' 
end from t1;

select case 
when dayname(now()) in ('Saturday','Sunday') then 'result undefined on weekends' 
when x > y then 'x greater than y' 
when x = y then 'x and y are equal' 
when x is null or y is null then 'one of the columns is null' 
else null end 
from t1;

Hive NVL conditional function examples

select nvl(null,'value is null');

Hive COALESCE conditional function examples

select coalesce(null,'a',null,'b');

Hive DECODE conditional function examples

SELECT event, decode(day_of_week, 1, "Monday", 2, "Tuesday", 3, "Wednesday", 4, "Thursday", 5, "Friday", 6, "Saturday", 7, "Sunday", "Unknown day") FROM calendar;

Spark DataFrame Union and UnionAll

Dataframe union() – union() method of the DataFrame is used to combine two DataFrame’s of the same structure/schema. If schemas are not the same it returns an error.

DataFrame unionAll() – unionAll() is deprecated since Spark “2.0.0” version and replaced with union().

Note: In other SQL’s, Union eliminates the duplicates but UnionAll combines two datasets including duplicate records. But, in spark both behave the same and use DataFrame duplicate function to remove duplicate rows.

import spark.implicits._

  val simpleData = Seq(("James","Sales","NY",90000,34,10000),
    ("Michael","Sales","NY",86000,56,20000),
    ("Robert","Sales","CA",81000,30,23000),
    ("Maria","Finance","CA",90000,24,23000)
  )
  val df = simpleData.toDF("employee_name","department","state","salary","age","bonus")
  df.printSchema()
  df.show()

This yields the below schema and DataFrame output.

root
 |-- employee_name: string (nullable = true)
 |-- department: string (nullable = true)
 |-- state: string (nullable = true)
 |-- salary: integer (nullable = false)
 |-- age: integer (nullable = false)
 |-- bonus: integer (nullable = false)

+-------------+----------+-----+------+---+-----+
|employee_name|department|state|salary|age|bonus|
+-------------+----------+-----+------+---+-----+
|        James|     Sales|   NY| 90000| 34|10000|
|      Michael|     Sales|   NY| 86000| 56|20000|
|       Robert|     Sales|   CA| 81000| 30|23000|
|        Maria|   Finance|   CA| 90000| 24|23000|
+-------------+----------+-----+------+---+-----+

Now, let’s create a second Dataframe with the new records and some records from the above Dataframe but with the same schema.

  val simpleData2 = Seq(("James","Sales","NY",90000,34,10000),
    ("Maria","Finance","CA",90000,24,23000),
    ("Jen","Finance","NY",79000,53,15000),
    ("Jeff","Marketing","CA",80000,25,18000),
    ("Kumar","Marketing","NY",91000,50,21000)
  )
  val df2 = simpleData2.toDF("employee_name","department","state","salary","age","bonus")

This yields below output

+-------------+----------+-----+------+---+-----+
|employee_name|department|state|salary|age|bonus|
+-------------+----------+-----+------+---+-----+
|James        |Sales     |NY   |90000 |34 |10000|
|Maria        |Finance   |CA   |90000 |24 |23000|
|Jen          |Finance   |NY   |79000 |53 |15000|
|Jeff         |Marketing |CA   |80000 |25 |18000|
|Kumar        |Marketing |NY   |91000 |50 |21000|
+-------------+----------+-----+------+---+-----+

Combine two or more DataFrames using union

DataFrame union() method combines two DataFrames and returns the new DataFrame with all rows from two Dataframes regardless of duplicate data.

  val df3 = df.union(df2)
  df3.show(false)

As you see below it returns all records.

+-------------+----------+-----+------+---+-----+
|employee_name|department|state|salary|age|bonus|
+-------------+----------+-----+------+---+-----+
|James        |Sales     |NY   |90000 |34 |10000|
|Michael      |Sales     |NY   |86000 |56 |20000|
|Robert       |Sales     |CA   |81000 |30 |23000|
|Maria        |Finance   |CA   |90000 |24 |23000|
|James        |Sales     |NY   |90000 |34 |10000|
|Maria        |Finance   |CA   |90000 |24 |23000|
|Jen          |Finance   |NY   |79000 |53 |15000|
|Jeff         |Marketing |CA   |80000 |25 |18000|
|Kumar        |Marketing |NY   |91000 |50 |21000|
+-------------+----------+-----+------+---+-----+

Combine DataFrames using unionAll

DataFrame unionAll() method is deprecated since Spark “2.0.0” version and recommends using the union() method.

  val df4 = df.unionAll(df2)
  df4.show(false)

Returns the same output as above.

Combine without Duplicates

Since the union() method returns all rows without distinct records, we will use the distinct() function to return just one record when duplicate exists.

  val df5 = df.union(df2).distinct()
  df5.show(false)

Yields below output. As you see, this returns only distinct rows.

+-------------+----------+-----+------+---+-----+
|employee_name|department|state|salary|age|bonus|
+-------------+----------+-----+------+---+-----+
|James        |Sales     |NY   |90000 |34 |10000|
|Maria        |Finance   |CA   |90000 |24 |23000|
|Jeff         |Marketing |CA   |80000 |25 |18000|
|Jen          |Finance   |NY   |79000 |53 |15000|
|Kumar        |Marketing |NY   |91000 |50 |21000|
|Michael      |Sales     |NY   |86000 |56 |20000|
|Robert       |Sales     |CA   |81000 |30 |23000|
+-------------+----------+-----+------+---+-----+

Complete Example of DataFrame Union

package com.sparkbyexamples.spark.dataframe

import org.apache.spark.sql.SparkSession

object UnionExample extends App{

  val spark: SparkSession = SparkSession.builder()
    .master("local[1]")
    .appName("SparkByExamples.com")
    .getOrCreate()

  spark.sparkContext.setLogLevel("ERROR")

  import spark.implicits._

  val simpleData = Seq(("James","Sales","NY",90000,34,10000),
    ("Michael","Sales","NY",86000,56,20000),
    ("Robert","Sales","CA",81000,30,23000),
    ("Maria","Finance","CA",90000,24,23000)
  )
  val df = simpleData.toDF("employee_name","department","state","salary","age","bonus")
  df.printSchema()
  df.show()

  val simpleData2 = Seq(("James","Sales","NY",90000,34,10000),
    ("Maria","Finance","CA",90000,24,23000),
    ("Jen","Finance","NY",79000,53,15000),
    ("Jeff","Marketing","CA",80000,25,18000),
    ("Kumar","Marketing","NY",91000,50,21000)
  )
  val df2 = simpleData2.toDF("employee_name","department","state","salary","age","bonus")
  df2.show(false)

  val df3 = df.union(df2)
  df3.show(false)
  df3.distinct().show(false)

  val df4 = df.unionAll(df2)
  df4.show(false)
}

Conclusion

In this Spark article, you have learned how to combine two or more DataFrame’s of the same schema into single DataFrame using Union method and learned the difference between the union() and unionAll() functions.