Apache Spark – Clear ur Doubt

Oct 27

Finding word counts in a file using Spark

Hey there! Welcome to ClearUrDoubt.com. In this post, we will look at the steps to find the word counts in a file using Spark 2.4.3. Input file:

Hey there!

Welcome to clearurdoubt.com. Happy Learning.

Hope you find these articles useful!

Happy Learning.

Hey there!

Welcome to clearurdoubt.com. Happy Learning.

Hope you find these articles useful!

Happy Learning.

Spark shell statements:

val rdd = spark.sparkContext.textFile("/clearurdoubt/test_file.txt")  // Read the input file
val words = rdd.flatMap(line => line.split(" ")).filter(word => !word.isEmpty)  // split the input lines to words and ignore the empty ones.
words.cache  // Cache the resource for future use

val totalWords = words.count  // total words in the input file

val pairs = words.map(word => (word, 1))  // parse the words into (word, noOfOccurence) pairs
val wordCounts = pairs.reduceByKey(_ + _).collect  // finding the each word count in terms of pairs and collecting them into an Array

val rdd = spark.sparkContext.textFile("/clearurdoubt/test_file.txt") // Read the input file

val words = rdd.flatMap(line => line.split(" ")).filter(word => !word.isEmpty) // split the input lines to words and ignore the empty ones.

words.cache // Cache the resource for future use

val totalWords = words.count // total words in the input file

val pairs = words.map(word => (word, 1)) // parse the words into (word, noOfOccurence) pairs

val wordCounts = pairs.reduceByKey(_ + _).collect // finding the each word count in terms of pairs and collecting them into an Array

Output:

totalWords: Long = 15

wordCounts: Array[(String, Int)] = Array((Learning.,2), (Hey,1), (Welcome,1), (Happy,2), (Hope,1), (find,1), (you,1), (there!,1), (to,1), (clearurdoubt.com.,1), (articles,1), (these,1), (useful!,1))

totalWords: Long = 15

wordCounts: Array[(String, Int)] = Array((Learning.,2), (Hey,1), (Welcome,1), (Happy,2), (Hope,1), (find,1), (you,1), (there!,1), (to,1), (clearurdoubt.com.,1), (articles,1), (these,1), (useful!,1))

Happy Learning! Please leave a reply Read more

Jun 09

How to add a new column and update its value based on the other column in the Dataframe in Spark

Hey there! Welcome to ClearUrDoubt.com. In this post, we will look at updating a column value based on another column value in a dataframe using when() utility function in Spark.

Spark context available as 'sc' (master = local[*], app id = local-1560028311690).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.3
      /_/

Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 12.0.1)
Type in expressions to have them evaluated.
Type :help for more information.

scala> val df = spark.read.json("/clearurdoubt/output.json");
df: org.apache.spark.sql.DataFrame = [first_name: string, id: bigint ... 2 more fields]

scala> df.show(false);
+----------+---+---------+------------+
|first_name|id |last_name|student_type|
+----------+---+---------+------------+
|Rahul     |101|Kumar    |1           |
|Antony    |102|James    |0           |
|Ashok     |103|Kedar    |0           |
|Ravi      |104|Aswin    |1           |
+----------+---+---------+------------+


scala> val new_df = df.withColumn("student_type_description", when(col("student_type").equalTo("1"), lit("Day-Scholar")).otherwise("Residential"));
new_df: org.apache.spark.sql.DataFrame = [first_name: string, id: bigint ... 3 more fields]

scala> new_df.show(false);
+----------+---+---------+------------+------------------------+
|first_name|id |last_name|student_type|student_type_description|
+----------+---+---------+------------+------------------------+
|Rahul     |101|Kumar    |1           |Day-Scholar             |
|Antony    |102|James    |0           |Residential             |
|Ashok     |103|Kedar    |0           |Hostler                 |
|Ravi      |104|Aswin    |1           |Residential             |
+----------+---+---------+------------+------------------------+


scala>

Spark context available as 'sc' (master = local[*], app id = local-1560028311690).

Spark session available as 'spark'.

Welcome to

____ __

/ __/__ ___ _____/ /__

_\ \/ _ \/ _ `/ __/ '_/

/___/ .__/\_,_/_/ /_/\_\ version 2.4.3

/_/

Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 12.0.1)

Type in expressions to have them evaluated.

Type :help for more information.

scala> val df = spark.read.json("/clearurdoubt/output.json");

df: org.apache.spark.sql.DataFrame = [first_name: string, id: bigint ... 2 more fields]

scala> df.show(false);

+----------+---+---------+------------+

+----------+---+---------+------------+

|Rahul |101|Kumar |1 |

|Antony |102|James |0 |

|Ashok |103|Kedar |0 |

|Ravi |104|Aswin |1 |

+----------+---+---------+------------+

scala> val new_df = df.withColumn("student_type_description", when(col("student_type").equalTo("1"), lit("Day-Scholar")).otherwise("Residential"));

new_df: org.apache.spark.sql.DataFrame = [first_name: string, id: bigint ... 3 more fields]

scala> new_df.show(false);

+----------+---+---------+------------+------------------------+

+----------+---+---------+------------+------------------------+

+----------+---+---------+------------+------------------------+

scala>

WHEN() function documentation: static Column when(Column condition, Object value) Evaluates a list Read more

Jun 05

How to add a new column to a Dataframe in Apache Spark

Hey there! Welcome to ClearUrDoubt.com. In this post, we will look at withColumn() function in Apache Spark SQL API. withColumn(String columnName, Column newColumn) is used to add a new column to a dataframe. Let’s look at the below code snippet in spark-shell for Read more

Jun 05

How to rename a column in a Dataframe in Apache Spark

Hey there! Welcome to ClearUrDoubt.com. In this post, we will look at withColumnRenamed() function in Apache Spark SQL API. withColumnRenamed(String columnName, String newColumnName) is used to rename a column in a Dataframe. Let’s look at the below code snippet in spark-shell for Read more

Oct 29

Spark program to load a table data from Cassandra to Hive using Java

Hey there! Welcome to ClearUrDoubt.com. In this post, we will look at a Spark program to load a table data from Cassandra to Hive using Java. Step 1: Create a table in Cassandra and insert records into it.

cqlsh:clearurdoubt> CREATE TABLE students(
                ... id int PRIMARY KEY,
                ... name text,
                ... marks int);

cqlsh:clearurdoubt> INSERT INTO students (id, name, marks) VALUES (101, 'Sidhu', 92);
cqlsh:clearurdoubt> INSERT INTO students (id, name, marks) VALUES (102, 'Sanju', 90);
cqlsh:clearurdoubt> INSERT INTO students (id, name, marks) VALUES (103, 'Ram', 45) ;
cqlsh:clearurdoubt> INSERT INTO students (id, name, marks) VALUES (105, 'Krishna', 88) ;
cqlsh:clearurdoubt> INSERT INTO students (id, name, marks) VALUES (104, 'Mounika', 90) ;
cqlsh:clearurdoubt>
cqlsh:clearurdoubt> select * from students;

 id  | marks | name
-----+-------+---------
 105 |    88 | Krishna
 104 |    90 | Mounika
 102 |    90 |   Sanju
 101 |    92 |   Sidhu
 103 |    45 |     Ram

(5 rows)
cqlsh:clearurdoubt>

cqlsh:clearurdoubt> CREATE TABLE students(

... id int PRIMARY KEY,

... name text,

... marks int);

cqlsh:clearurdoubt> INSERT INTO students (id, name, marks) VALUES (101, 'Sidhu', 92);

cqlsh:clearurdoubt> INSERT INTO students (id, name, marks) VALUES (102, 'Sanju', 90);

cqlsh:clearurdoubt> INSERT INTO students (id, name, marks) VALUES (103, 'Ram', 45) ;

cqlsh:clearurdoubt> INSERT INTO students (id, name, marks) VALUES (105, 'Krishna', 88) ;

cqlsh:clearurdoubt> INSERT INTO students (id, name, marks) VALUES (104, 'Mounika', 90) ;

cqlsh:clearurdoubt>

cqlsh:clearurdoubt> select * from students;

id | marks | name

-----+-------+---------

105 | 88 | Krishna

104 | 90 | Mounika

102 | 90 | Sanju

101 | 92 | Sidhu

103 | 45 | Ram

(5 rows)

cqlsh:clearurdoubt>

Step 2: Read more

Oct 14

Spark Program to load a CSV file into a Dataset using Java 8

Hey there! Welcome to ClearUrDoubt.com. In this post, we will look at a Spark(2.3.0) Program to load a CSV file into a Dataset using Java 8. Please go through the below post before going through this post. Program to load a Read more

Sep 27

Program to load a text file into a Dataset in Spark using Java 8

Hey there! Welcome to ClearUrDoubt.com. In this post, we will look at a program to load a text file into a Dataset in Spark(2.3.0) using Java 8. Check out the below classes/concepts before going through the program: static org.apache.spark.sql.functions.col org.apache.spark.sql.Dataset org.apache.spark.sql.Row Read more

Sep 26

Word Count program using Spark and Java 8

Hey there! Welcome to ClearUrDoubt.com. In this post, we will look at the WordCount program in Spark using Java 8. Before going through the below program, it’s better to look into below concepts for a better understanding of the program: org.apache.spark.api.java.JavaRDD org.apache.spark.api.java.JavaPairRDD Read more

Sep 23

How to read a CSV file in spark-shell using Spark SQL

Hey there! Welcome to ClearUrDoubt.com. In this post, we will go through the steps to read a CSV file in Spark SQL using spark-shell. Step 1: In Spark 1.6.0, to read a CSV file, we need to use a Read more

Aug 24

Word Count program using Spark and Scala

Hey there! Welcome to ClearUrDoubt.com. In this post, we will look at the WordCount program in Spark using Scala. Here is the Scala program:

package com.sample

import org.apache.spark._


object ScalaWordCounts {
  
  def main(args: Array[String])
  {
    val conf = new SparkConf().setAppName("Scala Word Count").setMaster("local")
    val sc = new SparkContext(conf)
  
    val lines = sc.textFile(args(0))
    val words = lines.flatMap(line => line.split(" "))
    val pairs = words.map(word => (word, 1))
    val counts = pairs.reduceByKey((count1, count2) => (count1 + count2))
  
    counts.saveAsTextFile(args(1))
    sc.stop()
  }
  
}

package com.sample

import org.apache.spark._

object ScalaWordCounts {

def main(args: Array[String])

{

val conf = new SparkConf().setAppName("Scala Word Count").setMaster("local")

val sc = new SparkContext(conf)

val lines = sc.textFile(args(0))

val words = lines.flatMap(line => line.split(" "))

val pairs = words.map(word => (word, 1))

val counts = pairs.reduceByKey((count1, count2) => (count1 + count2))

counts.saveAsTextFile(args(1))

sc.stop()

}

Input(HDFS) file:

[admin@myworkstation ~]$ hadoop fs -cat /user/workspace/clearurdoubt/test_folder1/testfile_1.txt
This is a test file.

it is created for the practice purpose.
[admin@myworkstation ~]$

[admin@myworkstation ~]$ hadoop fs -cat /user/workspace/clearurdoubt/test_folder1/testfile_1.txt

This is a test file.

it is created for the practice purpose.

[admin@myworkstation ~]$

Spark-submit command:

spark-submit --master local --class com.sample.ScalaWordCounts /user/workspace/clearurdoubt/lib/ScalaWordCount-0.0.1-SNAPSHOT.jar /user/workspace/clearurdoubt/test_folder1/testfile_1.txt /user/workspace/clearurdoubt/word_count_output/

1	spark-submit --master local --class com.sample.ScalaWordCounts /user/workspace/clearurdoubt/lib/ScalaWordCount-0.0.1-SNAPSHOT.jar /user/workspace/clearurdoubt/test_folder1/testfile_1.txt /user/workspace/clearurdoubt/word_count_output/

Output:

[admin@myworkstation ~]$ hdfs dfs -cat /user/workspace/clearurdoubt/word_count_output/*
(created,1)
(it,1)
(is,2)
(practice,1)
(a,1)
(,1)
(This,1)
(for,1)
(file.,1)
(purpose.,1)
(the,1)
(test,1)
[admin@myworkstation ~]$

[admin@myworkstation ~]$ hdfs dfs -cat /user/workspace/clearurdoubt/word_count_output/*

(created,1)

(it,1)

(is,2)

(practice,1)

(a,1)

(,1)

(This,1)

(for,1)

(file.,1)

(purpose.,1)

(the,1)

(test,1)

[admin@myworkstation ~]$

Happy Learning! 🙂 Please Read more