Spark Dataframe commands

Question 1

Describe a dataframe in your own words

Answer 1

Dataframe is like a table with rows and columns

Question 2

How do you read a table from Hive into Spark dataframe with select statement- Spark 2.6

Answer 2

spark.sql(“Select * from db.mytable”)

Question 3

How do you read a table from Hive into Spark dataframe without select statement- Spark 2.6

Answer 3

spark.table(“db.myTable”)

Question 4

How to display a dataframe

Answer 4

df.show()

Question 5

How to display exactly 100 rows of a dataframe

Answer 5

df.show(100)

Question 6

Why do we pass True/False in show

Answer 6

True expands the columns and False compresses the columns during show

Question 7

Select specific columns

Answer 7

df.select(‘col1’,’col2’,’col3’)

Question 8

Can I pass a list of columns within the select statement

Answer 8

Yes. df.select([‘col1’, ‘col2’,’col3’])

Question 9

How do I change the column name without using withColumnRenamed?

Answer 9

df.selectExpr(“col1”,”col2 as test_column”)

Question 10

How do I pull specific rows from a dataframe - For example where a certain column in my dataframe is true

Answer 10

df.filter(“col1 = True”)

Question 11

import statement to import functions

Answer 11

from pyspark.sql import functions as func

Question 12

Get the total number of records in a dataframe

Answer 12

df.count()

Question 13

How do I get the count of distinct values in a column?

Answer 13

df.dropDuplicates(“col1”).count()

Question 14

What is the difference between df.dropDuplicates() and df.dropDuplicates(“col1”)

Answer 14

dropDuplicates() drops the duplicates in the entire dataframe and dropDuplicates(“col1”) just drops the duplicates in specific column

Question 15

How do I see the schema of a dataframe

Answer 15

df.printSchema()

Question 16

How do I see the column names along with the datatypes

Answer 16

df.printSchema()

Question 17

How do I retrieve the columns to a python list

Answer 17

df.columns

Question 18

df.columns() - Is this correct and what will be the output

Answer 18

No. The braces shouldnt be present. It throws an error

Question 19

How do I drop a column from a dataframe?

Answer 19

df.drop(“col1”)

Question 20

df.drop([“col1”,”col2”]) - Is this correct and why

Answer 20

yes it is correct. We can pass a list in drop function

Question 21

Groupby syntax with count

Answer 21

df.groupBy(“col1”).agg(func.count(“col2”))

Question 22

Order the rows in a dataframe on a certain column.

Answer 22

df.orderBy(func.asc(“col1”))

Question 23

Groupby on multiple columns syntax with count

Answer 23

df.groupBy(“col1”,”col2”).agg(func.count(“col2”))

Question 24

Order the rows in a dataframe on multiple columns.

Answer 24

df.orderBy(func.asc(“col1”), func.desc(“col2”))

Question 25

case expression general syntax

Answer 25

case when col1 = 'Y' then 'True' when col1 = 'N' then 'False' else 'NA' end

Question 26

drop multiple columns

Answer 26

df.drop("col1","col2")

Question 27

drop duplicate values in multiple columns

Answer 27

dropDuplicates(["col1", "col2"])

Question 28

Create a new column in the dataframe. The new column is a flag that has true or false. If a column value is > 100 then True else false

Answer 28

df.withColumn("flag", func.expr("case when col1 >= 100 then True else False end"))

Question 29

I have a dataframe with some records. I need to flag all the records as 'True' before I proceed further. How do I do that

Answer 29

df.withColumn("flag", func.lit(True))

Question 30

Rename a column

Answer 30

df.withColumnRenamed("old_col_name", "new_col_name")

Question 31

Two ways to rename a column

Answer 31

withColumnRenamed, selectExpr

Question 32

Ways to create dataframe

Answer 32

1. Reading hive tables 2. Reading CSV or JSON files 3. Create dataframe from list 4. Create dataframe from rdd

Question 33

How to read csv files into dataframe?

Answer 33

1. df = spark.read.csv("file.csv") | 2. df = spark.read.format("csv").load("file.csv")

Question 34

Column names for this dataframe - df = spark.read.format("csv").load("file.csv")

Answer 34

_c0, _c1, _c2...

Question 35

How to load header for csv read command

Answer 35

``` df2 = spark.read.option("header",True).csv("file.csv") df2 = spark.read.options(header = 'True').csv("file.csv") ```

Question 36

PySpark reads all columns as a ________ data type by default

Answer 36

string

Question 37

Read multiple csv files into a single dataframe

Answer 37

df = spark.read.csv("path1,path2,path3")

Question 38

Read all CSV files from a directory into DataFrame

Answer 38

df = spark.read.csv("Folder path")

Question 39

Specify a specific delimiter while reading csv

Answer 39

``` df3 = spark.read.option("delimiter",",") .csv("test.csv") df3 = spark.read.options(delimiter=',') .csv("test.csv") ```

Question 40

How to change the default datatype read by spark from a csv

Answer 40

``` df3 = spark.read.option("inferschema", True) .csv("test.csv") df3 = spark.read.options(inferschema='True') .csv("test.csv") ```

Question 41

Set both delimiter and inferschema

Answer 41

df3 = spark.read.option("delimiter",",").option("inferschema",True) .csv("test.csv") df3 = spark.read.options(inferschema='True', delimiter = '|') .csv("test.csv")

Question 42

how to import datatypes

Answer 42

from pyspark.sql.types import *

Question 43

Read custom schema - I don't want default string schema and also I don't want inferschema but would like to change to custom datatype

Answer 43

from pyspark.sql.types import * from pyspark.sql.types import StructType,StructField, StringType, IntegerType schema = StructType([ StructField('firstname', StringType(), True), StructField('middlename', StringType(), True), StructField('id', IntegerType(), True) ]) df = spark.read.format("csv").option("Header", True).schema(schema).load("file.csv") df = spark.read.option("Header", True).schema(schema).csv("file.csv")

Question 44

Write a dataframe to csv file with no header

Answer 44

df. write.format("csv").option("header", True).save("demo.csv") df. write.option("header", True).csv("demo.csv")

Question 45

Modes while saving a dataframe as a file

Answer 45

1. overwrite – mode is used to overwrite the existing file. 2. append – To add the data to the existing file. 3. ignore – Ignores write operation when the file already exists. 4. error – This is a default option when the file already exists, it returns an error.

Question 46

append mode

Answer 46

df.write.mode("append").option("header", True).csv("demo.csv")

Question 47

create spark session

Answer 47

spark = SparkSession \ .builder \ .appName("App1") \ .getOrCreate()

Question 48

Check the type of variable

Answer 48

type(df)

Question 49

Read a json file

Answer 49

1. df = spark.read.json("file.json") | 2. df = spark.read.format("json").load("file.json")

Question 50

import statement to import SparkSession

Answer 50

from pyspark.sql import SparkSession

Question 51

``` [{ "RecordNumber": 2, "Zipcode": 704, "ZipCodeType": "STANDARD", "City": "PASEO COSTA DEL SUR", "State": "PR" }, { "RecordNumber": 10, "Zipcode": 709, "ZipCodeType": "STANDARD", "City": "BDA SAN LUIS", "State": "PR" }] ``` Read the multiline json records

Answer 51

df = spark.read.options(mutliline="True").json("file.json")

Question 52

Read multiple json files

Answer 52

df = spark.read.json(["json path1","json path2","json path3"])

Question 53

Read all json files in a directory

Answer 53

df = spark.read.json("files/*.json")

Question 54

Pass custom schema for each of the columns for json

Answer 54

from pyspark.sql.types import * from pyspark.sql.types import StructType,StructField, StringType schema = StructType([ StructField('firstname', StringType(), True), StructField('middlename', StringType(), True), StructField('lastname', StringType(), True) ]) df = spark.read.options(header = 'True').schema(schema).json("file.json")

Question 55

Write a dataframe as json file

Answer 55

df.write.json("file.json")

Question 56

Write a dataframe as json file - append mode

Answer 56

df.write.mode("append").json("file.json")

Question 57

Create a dataframe using parallelize

Answer 57

from pyspark.sql import Row dept = [Row("A",10), Row("B",20), Row("C",30)] ``` rdd = spark.sparkContext.parallelize(dept) df = rdd.toDF(col1, col2) ```

Question 58

Create dataframe from list without using parallelize

Answer 58

``` dept = [("A",10), ("B",20), ("C",30)] col_names= ("col_1_name", "col_2_name") df = spark.createDataFrame(data = dept, schema = col_names) ```

Question 59

I have two tables. One table has id and location. The second table has all the ids who are assigned a parking space . I need a output report of the id, location and whether parking space is allocated - is_parking_allocated (Y/N)

Answer 59

join

Question 60

I have two tables. Table 1 - ids, location; Table 2 - ids salary. Output: ids, location and salary

Answer 60

join

Question 61

I have two tables. Table 1 - Parts and price Table 2 - Only the parts are purchased in the last 2 months. Output: Parts, price and a flag if the parts are purchased in the last 2 monts - flag_2_months

Answer 61

join

Question 62

Declare udf

Answer 62

from pyspark.sql import functions as func ``` def split_str("s"): return s.split("_")[1] ``` split_str_udf = func.udf(split_str) #udf registration df1 = df.withColumn("last_name", split_str("full_name"))

Question 63

Read avro

Answer 63

df = spark.read.format("avro").load("avro_file_path")

Question 64

save avro

Answer 64

df.write.format("avro").save("avro_file_path")

Question 65

left outer join df1 with df2 with alias; join on id present in both tables and select two columns, one from each table

Answer 65

df1.alias("a").join(df2.alias("b"), df1.id == df2.id, "left_outer").select("a.col1","b.col2")

Question 66

inner join df1 with df2 with alias; join on id present in both tables and select two columns, one from each table

Answer 66

df1.alias("a").join(df2.alias("b"), df1.id == df2.id, "inner").select("a.col1","b.col2")

Question 67

right outer join df2 with df1 with alias; join on id present in both tables and select two columns, one from each table

Answer 67

df2.alias("a").join(df1.alias("b"), df2.id == df1.id, "right_outer").select("df1.col1","df2.col2")