How to append content to a DBFS file using python spark

You can read and write to DBFS files using 'dbutils'

Lets see one example

dbutils.fs.put("dbfs:///mnt/sample.txt", "sample content")

Above command helps to write "sample content" to 'dbfs:///mnt/sample.txt'

Now you have the file in your DBFS location. Lets see how to append content to this file.

In order to do that, you can open the file in append mode.

with  open("/dbfs/mnt/sample.txt", "a") as f:
  f.write("append values")

Now your appended file is ready!!!

 

How to access data in Delta tables

Delta tables can be accessed either by specifying the path on DBFS or by table name.

You can check my previous blog to see how to write delta files here.  I will be using the same example location here.

Option 1 : Read delta tables by specifying DBFS path

val employeeDF = spark.read.format("delta").load("/mnt/delta/Employee")

Option 2 : Read delta table by  table name

val employeeDF = spark.table("employee")

Create Delta table on csv file in python spark

 

You can read files into Dataframe and write out in delta format

Step 1 : Read the input csv

Step 2 : Write the csv to ADLS location using Delta format

Step 3: Create a table on top of it

myCSV= spark.read.csv("/path/to/input/data",header=True,sep=","); 
myCSV.write.format("delta").mode("overwrite").option('overwriteSchema','true').save("/mnt/delta/Employee") 
spark.sql("CREATE TABLE employee USING DELTA LOCATION '/mnt/delta/Employee/'") 

Create Empty Dataframe In Spark - Without Defining Schema

 

There may be cases where we need to initialize a Dataframe without specifying a schema.
Let's see how to do that

empty_DF = sqlContext.createDataFrame(sc.emptyRDD(),StructType([]))

StructType([]) This creates an empty schema for our dataframe.

Create Widget in Databricks Python Notebook

 In order to get some inputs from user we will require widgets in our Azure Databricks notebook.
This blog helps you to create a text based widget in your python notebook.

Syntax 

dbutils.widgets.text(<WidgetID>,<DefaultValue>,<DisplayName>)

Lets see the result of above widget in Notebook.

Oh yeah....our widget got created.
Now lets enter some inputs in the widget.

Now if you check, the code block has already executed and displayed your entered widget value.

You are done.

Go ahead and create one 😊

Pivot a DataFrame

How to pivot the data to create multiple columns out of 1 column with multiple rows.

spark-shell --queue= *;

To adjust logging level use sc.setLogLevel(newLevel).
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 1.6.0
Spark context available as sc 
SQL context available as sqlContext.

scala>  val sqlcontext = new org.apache.spark.sql.SQLContext(sc)
sqlcontext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@4f9a8d71  

scala> val BazarDF = Seq(
     | ("Veg", "tomato", 1.99),
     | ("Veg", "potato", 0.45),
     | ("Fruit", "apple", 0.99),
     | ("Fruit", "pineapple", 2.59),
     | ("Fruit", "apple", 1.99)
     | ).toDF("Type", "Item", "Price")
BazarDF: org.apache.spark.sql.DataFrame = [Type: string, Item: string, Price: double]

scala> BazarDF.show()
+-----+---------+-----+
| Type|     Item|Price|
+-----+---------+-----+
|  Veg|   tomato| 1.99|
|  Veg|   potato| 0.45|
|Fruit|    apple| 0.99|
|Fruit|pineapple| 2.59|
|Fruit|    apple| 1.99|
+-----+---------+-----+

A pivot can be thought of as translating rows into columns while applying one or more aggregations.
Lets see how we can achieve the same using the above dataframe.
We will pivot the data based on "Item" column.

scala> BazarDF.groupBy("Type").pivot("Item").agg(min("Price")).show()
+-----+-----+---------+------+------+
| Type|apple|pineapple|potato|tomato|
+-----+-----+---------+------+------+
|  Veg| null|     null|  0.45|  1.99|
|Fruit| 0.99|     2.59|  null|  null|
+-----+-----+---------+------+------+

You are Done!

How to select multiple columns from a spark data frame using List[Column]

Let us create Example DataFrame to explain how to select List of columns of type "Column" from a dataframe

spark-shell --queue= *;

To adjust logging level use sc.setLogLevel(newLevel).
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 1.6.0
Spark context available as sc 
SQL context available as sqlContext.

scala>  val sqlcontext = new org.apache.spark.sql.SQLContext(sc)
sqlcontext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@4f9a8d71  

scala> val BazarDF = Seq(
     | ("Veg", "tomato", 1.99),
     | ("Veg", "potato", 0.45),
     | ("Fruit", "apple", 0.99),
     | ("Fruit", "pineapple", 2.59),
     | ("Fruit", "apple", 1.99)
     | ).toDF("Type", "Item", "Price")
BazarDF: org.apache.spark.sql.DataFrame = [Type: string, Item: string, Price: double]

scala> BazarDF.show()
+-----+---------+-----+
| Type|     Item|Price|
+-----+---------+-----+
|  Veg|   tomato| 1.99|
|  Veg|   potato| 0.45|
|Fruit|    apple| 0.99|
|Fruit|pineapple| 2.59|
|Fruit|    apple| 1.99|
+-----+---------+-----+

Create a List[Column] with column names.

scala> var selectExpr : List[Column] = List("Type","Item","Price")
<console>:25: error: not found: type Column
         var selectExpr : List[Column] = List("Type","Item","Price")
                               ^

If you are getting the same error Please take a look into this page .

Using : _* annotation select the columns from dataframe.

scala> var dfNew = BazarDF.select(selectExpr: _*)
dfNew: org.apache.spark.sql.DataFrame = [Type: string, Item: string, Price: double]

scala> dfNew.show()
+-----+---------+-----+
| Type|     Item|Price|
+-----+---------+-----+
|  Veg|   tomato| 1.99|
|  Veg|   potato| 0.45|
|Fruit|    apple| 0.99|
|Fruit|pineapple| 2.59|
|Fruit|    apple| 1.99|
+-----+---------+-----+

You are Done!


I have also explained How to select multiple columns from a sparkdata frame using List[String]