Skip to content

VinayChaudhari1996/pyspark-dataframe-made-easy

BranchesTags

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

13 Commits
SPARK
SPARK
 
 
README.md
README.md
 
 

Repository files navigation

Pyspark Dataframe made easy 😀

(Zero to Hero)

  • Vinay Chaudhari

Before you start,

  1. Open Google colab or Any IDE and hit following cmd.
  2. pip install pyspark

A

○ agg

○ alias

○ agg

C

○ cache

○ coalesce

○ columns

○ corr

○ count

○ cov

○ crosstab

○ cube

○ coalesce

D

○ describe

○ destinct

○ drop

○ dropDuplicates

○ dropna

○ dtypes

E

○ explain

F

○ fillna

○ filter

○ first

○ flatmap

○ foreach

○ foreachPartition

○ freqItems

G

○ groupBy

H

○ head

I

○ intersect

○ isLocal

J

○ join

L

○ limit

M

○ map

○ mapPartitions

N

○ na

O

○ orderBy

P

○ persist

○ printSchema

R

○ randomSplit

○ rdd

○ registerTempTable

○ repartition

○ replace

○ rollup

S

○ sample

○ sampleBy

○ schema

○ select

○ selectExpr

○ show

○ sort

○ sortWithPartitions

○ stat

○ subtract

✔ CONVERSIONS

T

○ take

○ toDF

○ toJSON

○ toPANDAS

U

○ unionAll

○ upersist

W

○ where(filter)

○ withColumn

○ withColumnRenamed

○ write

**** MAKE PR FOR CONTRIBUTION AND SUGGESTIONS :) ****

import IPython 
#agg

x = sqlContext.createDataFrame([("vinay","sunny",100),("deepak","parag",200),("akash","pravin",300)],['from','to','amount'])
y = x.agg({"amount":"avg"})

x.show()
y.show()
+------+------+------+
|  from|    to|amount|
+------+------+------+
| vinay| sunny|   100|
|deepak| parag|   200|
| akash|pravin|   300|
+------+------+------+

+-----------+
|avg(amount)|
+-----------+
|      200.0|
+-----------+

#alias
from pyspark.sql.functions import col
x = sqlContext.createDataFrame([("vinay","sunny",100),("deepak","parag",200),("akash","pravin",300)],['from','to','amount'])
y = x.alias("transactions")

x.show()
y.select(col("transactions.to")).show()
+------+------+------+
|  from|    to|amount|
+------+------+------+
| vinay| sunny|   100|
|deepak| parag|   200|
| akash|pravin|   300|
+------+------+------+

+------+
|    to|
+------+
| sunny|
| parag|
|pravin|
+------+

#cache 

x = sqlContext.createDataFrame([("vinay","sunny",100),("deepak","parag",200),("akash","pravin",300)],['from','to','amount'])
x.cache()

print(x.count()) #first action materializes x in memory
print(x.count()) #later actions avoid IO overhead
3
3

#coalesce

x_rdd = sc.parallelize([("vinay","sunny",100),("deepak","parag",200),("akash","pravin",300)],4)
x = sqlContext.createDataFrame(x_rdd,['from','to','amount'])
y = x.coalesce(numPartitions=1)

print(x.rdd.getNumPartitions())
print(y.rdd.getNumPartitions())

x.show()
y.show()
4
1
+------+------+------+
|  from|    to|amount|
+------+------+------+
| vinay| sunny|   100|
|deepak| parag|   200|
| akash|pravin|   300|
+------+------+------+

+------+------+------+
|  from|    to|amount|
+------+------+------+
| vinay| sunny|   100|
|deepak| parag|   200|
| akash|pravin|   300|
+------+------+------+

#collect

x = sqlContext.createDataFrame([("vinay","sunny",100),("deepak","parag",200),("akash","pravin",300)],['from','to','amount'])
y = x.collect() # it creates list of rows.
x.show()
print(y)
+------+------+------+
|  from|    to|amount|
+------+------+------+
| vinay| sunny|   100|
|deepak| parag|   200|
| akash|pravin|   300|
+------+------+------+

[Row(from='vinay', to='sunny', amount=100), Row(from='deepak', to='parag', amount=200), Row(from='akash', to='pravin', amount=300)]

#columns 

x = sqlContext.createDataFrame([("vinay","sunny",100),("deepak","parag",200),("akash","pravin",300)],['from','to','amount'])

y = x.columns
x.show()
print(y)
+------+------+------+
|  from|    to|amount|
+------+------+------+
| vinay| sunny|   100|
|deepak| parag|   200|
| akash|pravin|   300|
+------+------+------+

['from', 'to', 'amount']

#corr : Calculates the correlation of
# two columns of a DataFrame as a double value. 

x = sqlContext.createDataFrame([("vinay","sunny",100,300),("deepak","parag",200,600),("akash","pravin",300,900)], ['from','to','amount','fees'])
y = x.corr(col1="amount",col2="fees")
x.show()
print(y)
+------+------+------+----+
|  from|    to|amount|fees|
+------+------+------+----+
| vinay| sunny|   100| 300|
|deepak| parag|   200| 600|
| akash|pravin|   300| 900|
+------+------+------+----+

1.0

#count 
#Returns the number of rows in this DataFrame.
x = sqlContext.createDataFrame([("vinay","sunny",100),("deepak","parag",200),("akash","pravin",300)],['from','to','amount'])
x.show()
print(x.count())
+------+------+------+
|  from|    to|amount|
+------+------+------+
| vinay| sunny|   100|
|deepak| parag|   200|
| akash|pravin|   300|
+------+------+------+

3

#cov
#Calculate the sample covariance for the given columns,
#specified by their names, as a double value. 

x = sqlContext.createDataFrame([("vinay","sunny",100,300),("deepak","parag",200,600),("akash","pravin",300,900)], ['from','to','amount','fees'])
y = x.cov(col1="amount",col2="fees")

x.show()
print(y)
+------+------+------+----+
|  from|    to|amount|fees|
+------+------+------+----+
| vinay| sunny|   100| 300|
|deepak| parag|   200| 600|
| akash|pravin|   300| 900|
+------+------+------+----+

30000.0

#crosstab
x = sqlContext.createDataFrame([("vinay","deepak",0.1),("sunny","pratik",0.2),("parag","akash",0.3)], ['from','to','amt'])
y = x.crosstab(col1='from',col2='to')
x.show()
y.show()
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|0.1|
|sunny|pratik|0.2|
|parag| akash|0.3|
+-----+------+---+

+-------+-----+------+------+
|from_to|akash|deepak|pratik|
+-------+-----+------+------+
|  parag|    1|     0|     0|
|  vinay|    0|     1|     0|
|  sunny|    0|     0|     1|
+-------+-----+------+------+

col1 – The name of the first column. Distinct items will make the first item of each row.

col2 – The name of the second column. Distinct items will make the column names of the DataFrame.

#cube

# Create a multi-dimensional cube for the current DataFrame using the specified columns,
# so we can run aggregation on them

x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3)], ['from','to','amt'])

y = x.cube('from','to')
x.show()
print(y)
y.sum().show()
y.max().show()
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
+-----+------+---+

<pyspark.sql.group.GroupedData object at 0x000001D968C4A320>
+-----+------+--------+
| from|    to|sum(amt)|
+-----+------+--------+
| null| akash|       3|
| null|  null|       6|
|vinay|deepak|       1|
|vinay|  null|       1|
| null|deepak|       1|
|parag| akash|       3|
| null|pratik|       2|
|parag|  null|       3|
|sunny|  null|       2|
|sunny|pratik|       2|
+-----+------+--------+

+-----+------+--------+
| from|    to|max(amt)|
+-----+------+--------+
| null| akash|       3|
| null|  null|       3|
|vinay|deepak|       1|
|vinay|  null|       1|
| null|deepak|       1|
|parag| akash|       3|
| null|pratik|       2|
|parag|  null|       3|
|sunny|  null|       2|
|sunny|pratik|       2|
+-----+------+--------+

# Describe 

x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3)], ['from','to','amt'])

x.show()
x.describe().show()
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
+-----+------+---+

+-------+-----+------+---+
|summary| from|    to|amt|
+-------+-----+------+---+
|  count|    3|     3|  3|
|   mean| null|  null|2.0|
| stddev| null|  null|1.0|
|    min|parag| akash|  1|
|    max|vinay|pratik|  3|
+-------+-----+------+---+

# Distinct 

x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3),("parag","akash",3),("parag","akash",3)], ['from','to','amt'])
y = x.distinct()

x.show()
y.show()
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
|parag| akash|  3|
|parag| akash|  3|
+-----+------+---+

+-----+------+---+
| from|    to|amt|
+-----+------+---+
|sunny|pratik|  2|
|vinay|deepak|  1|
|parag| akash|  3|
+-----+------+---+

# Drop 

x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3)], ['from','to','amt'])
y = x.drop('amt')

x.show()
y.show()
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
+-----+------+---+

+-----+------+
| from|    to|
+-----+------+
|vinay|deepak|
|sunny|pratik|
|parag| akash|
+-----+------+

# dropDuplicates

x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3),("parag","akash",3),("parag","akash",3)], ['from','to','amt'])
y = x.dropDuplicates(subset=['from','to'])

x.show()
y.show()
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
|parag| akash|  3|
|parag| akash|  3|
+-----+------+---+

+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
+-----+------+---+

#dropna 
x = sqlContext.createDataFrame([(None,"vinay",0.1),("vinay","sunny",None),("Peter",None,0.3),("Mark","Steve",0.2)], ['from','to','amount'])
y = x.dropna(how='any',subset=['from','to'])
x.show()
y.show()
+-----+-----+------+
| from|   to|amount|
+-----+-----+------+
| null|vinay|   0.1|
|vinay|sunny|  null|
|Peter| null|   0.3|
| Mark|Steve|   0.2|
+-----+-----+------+

+-----+-----+------+
| from|   to|amount|
+-----+-----+------+
|vinay|sunny|  null|
| Mark|Steve|   0.2|
+-----+-----+------+

#dtypes

x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3),("parag","akash",3),("parag","akash",3)], ['from','to','amt'])
y = x.dtypes

x.show()
print(y)
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
|parag| akash|  3|
|parag| akash|  3|
+-----+------+---+

[('from', 'string'), ('to', 'string'), ('amt', 'bigint')]

#Explain

x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3)], ['from','to','amt'])
x.show()

x.agg({"amt":"avg"}).explain(extended = True)

+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
+-----+------+---+

== Parsed Logical Plan ==
'Aggregate ['avg(amt#169L) AS avg(amt)#187]
+- AnalysisBarrier
      +- LogicalRDD [from#167, to#168, amt#169L], false

== Analyzed Logical Plan ==
avg(amt): double
Aggregate [avg(amt#169L) AS avg(amt)#187]
+- LogicalRDD [from#167, to#168, amt#169L], false

== Optimized Logical Plan ==
Aggregate [avg(amt#169L) AS avg(amt)#187]
+- Project [amt#169L]
   +- LogicalRDD [from#167, to#168, amt#169L], false

== Physical Plan ==
*(2) HashAggregate(keys=[], functions=[avg(amt#169L)], output=[avg(amt)#187])
+- Exchange SinglePartition
   +- *(1) HashAggregate(keys=[], functions=[partial_avg(amt#169L)], output=[sum#192, count#193L])
      +- *(1) Project [amt#169L]
         +- Scan ExistingRDD[from#167,to#168,amt#169L]

#fillna 

x = sqlContext.createDataFrame([(None,"deepak",1),("sunny",None,2),("parag",None,3)], ['from','to','amt'])
y = x.fillna(value = '---',subset = ['from','to'])

x.show()
y.show()
+-----+------+---+
| from|    to|amt|
+-----+------+---+
| null|deepak|  1|
|sunny|  null|  2|
|parag|  null|  3|
+-----+------+---+

+-----+------+---+
| from|    to|amt|
+-----+------+---+
|  ---|deepak|  1|
|sunny|   ---|  2|
|parag|   ---|  3|
+-----+------+---+

Filter (Most used api)

# Filter 

x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3)], ['from','to','amt'])
y = x.filter("amt > 2 ")

x.show()
y.show()
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
+-----+------+---+

+-----+-----+---+
| from|   to|amt|
+-----+-----+---+
|parag|akash|  3|
+-----+-----+---+

# First

x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3)], ['from','to','amt'])
y = x.first()

x.show()
print(y)
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
+-----+------+---+

Row(from='vinay', to='deepak', amt=1)

Foreach

# foreach
from __future__ import print_function

# setup
fn = './foreachExampleDataFrames.txt' 
open(fn, 'w').close()  # clear the file
def fappend(el,f):
    '''appends el to file f'''
    print(el,file=open(f, 'a+') )

# example
x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3)], ['from','to','amt'])

y = x.foreach(lambda x: fappend(x,fn)) # writes into foreachExampleDataFrames.txt
x.show() # original dataframe
print(y) # foreach returns 'None'
# print the contents of the file
with open(fn, "r") as foreachExample:
    print (foreachExample.read())
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
+-----+------+---+

None
Row(from='vinay', to='deepak', amt=1)
Row(from='sunny', to='pratik', amt=2)

# foreachPartition
from __future__ import print_function

# setup
fn = './foreachExampleDataFrames.txt' 
open(fn, 'w').close()  # clear the file
def fappend(el,f):
    '''appends el to file f'''
    print(el,file=open(f, 'a+') )

# example
x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3)], ['from','to','amt'])

y = x.foreach(lambda x: fappend(x,fn)) # writes into foreachExampleDataFrames.txt
x.show() # original dataframe
print(y) # foreach returns 'None'
# print the contents of the file
with open(fn, "r") as foreachExample:
    print (foreachExample.read())
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
+-----+------+---+

None
Row(from='parag', to='akash', amt=3)
Row(from='sunny', to='pratik', amt=2)
Row(from='vinay', to='deepak', amt=1)

# freqItems 

x = sqlContext.createDataFrame([("Vinay","sunny",50), \
                                ("Deepak","sunny",30), \
                                ("Vinay","Parag",20), \
                                ("Vinay","ram",50), \
                                ("sham","sunny",90), \
                                ("Vinay","pushpak",50), \
                                ("om","sunny",50), \
                                ("sagar","sunny",50), \
                                ("Vinay","rahul",80), \
                                ("akash","sunny",50), \
                                ("puranik","pranav",70)],\
                                ['from','to','amount'])

y = x.freqItems(cols=['from','amount'],support=0.8)

x.show()
y.show()
+-------+-------+------+
|   from|     to|amount|
+-------+-------+------+
|  Vinay|  sunny|    50|
| Deepak|  sunny|    30|
|  Vinay|  Parag|    20|
|  Vinay|    ram|    50|
|   sham|  sunny|    90|
|  Vinay|pushpak|    50|
|     om|  sunny|    50|
|  sagar|  sunny|    50|
|  Vinay|  rahul|    80|
|  akash|  sunny|    50|
|puranik| pranav|    70|
+-------+-------+------+

+--------------+----------------+
|from_freqItems|amount_freqItems|
+--------------+----------------+
|       [Vinay]|            [50]|
+--------------+----------------+

groupBy (most used api)

# groupBy

x = sqlContext.createDataFrame([("vinay","deepak",1),("sunny","pratik",2),("parag","akash",3)], ['from','to','amt'])
y = x.groupBy('amt')

x.show()
print(y)
+-----+------+---+
| from|    to|amt|
+-----+------+---+
|vinay|deepak|  1|
|sunny|pratik|  2|
|parag| akash|  3|
+-----+------+---+

<pyspark.sql.group.GroupedData object at 0x0000021742513CC0>

# groupBy (col1).avg(col2)

x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455)], ['from','to','amt'])
y = x.groupBy('from').avg('amt')

x.show()
y.show()
+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|sunny|pratik|  451232|
|parag| akash| 2555455|
+-----+------+--------+

+-----+-----------+
| from|   avg(amt)|
+-----+-----------+
|parag|  2555455.0|
|sunny|   451232.0|
|vinay|1.2466641E7|
+-----+-----------+

# head

x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455)], ['from','to','amt'])

y = x.head(2)
x.show()
print(y)
+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|sunny|pratik|  451232|
|parag| akash| 2555455|
+-----+------+--------+

[Row(from='vinay', to='deepak', amt=12466641), Row(from='sunny', to='pratik', amt=451232)]

# intersect

x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455),("parag","akash",2555455)], ['from','to','amt'])

y = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455),("parag","akashay",2555455)], ['from','to','amt'])

z = x.intersect(y)

x.show()
y.show()
z.show()
+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|sunny|pratik|  451232|
|parag| akash| 2555455|
|parag| akash| 2555455|
+-----+------+--------+

+-----+-------+--------+
| from|     to|     amt|
+-----+-------+--------+
|vinay| deepak|12466641|
|sunny| pratik|  451232|
|parag|  akash| 2555455|
|parag|akashay| 2555455|
+-----+-------+--------+

+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|sunny|pratik|  451232|
|vinay|deepak|12466641|
|parag| akash| 2555455|
+-----+------+--------+

# isLocal

x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455),("parag","akash",2555455)], ['from','to','amt'])


y = x.isLocal()

x.show()
print(y)
+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|sunny|pratik|  451232|
|parag| akash| 2555455|
|parag| akash| 2555455|
+-----+------+--------+

False

join (Most used api)

# join 
x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455),("Salman","akash",2555455)], ['from','to','amt'])
y = sqlContext.createDataFrame([('Andy',20),("Steve",40),("Elon",80)], ['name','age'])
z = x.join(y,x.to ==y.name,'inner').select('from','to','amt','age')
x.show()
y.show()
z.show()
+------+------+--------+
|  from|    to|     amt|
+------+------+--------+
| vinay|deepak|12466641|
| sunny|pratik|  451232|
| parag| akash| 2555455|
|Salman| akash| 2555455|
+------+------+--------+

+-----+---+
| name|age|
+-----+---+
| Andy| 20|
|Steve| 40|
| Elon| 80|
+-----+---+

+----+---+---+---+
|from| to|amt|age|
+----+---+---+---+
+----+---+---+---+

# join 
x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455),("Salman","akash",2555455)], ['from','to','amt'])
y = sqlContext.createDataFrame([('Andy',20),("Steve",40),("Elon",80)], ['name','age'])
z = x.join(y,x.to ==y.name,'outer').select('from','to','amt','age')
x.show()
y.show()
z.show()
+------+------+--------+
|  from|    to|     amt|
+------+------+--------+
| vinay|deepak|12466641|
| sunny|pratik|  451232|
| parag| akash| 2555455|
|Salman| akash| 2555455|
+------+------+--------+

+-----+---+
| name|age|
+-----+---+
| Andy| 20|
|Steve| 40|
| Elon| 80|
+-----+---+

+------+------+--------+----+
|  from|    to|     amt| age|
+------+------+--------+----+
|  null|  null|    null|  40|
| sunny|pratik|  451232|null|
| vinay|deepak|12466641|null|
|  null|  null|    null|  20|
| parag| akash| 2555455|null|
|Salman| akash| 2555455|null|
|  null|  null|    null|  80|
+------+------+--------+----+

# Limit

# join 
x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455),("Salman","akash",2555455)], ['from','to','amt'])
y = x.limit(2)
x.show()
y.show()
+------+------+--------+
|  from|    to|     amt|
+------+------+--------+
| vinay|deepak|12466641|
| sunny|pratik|  451232|
| parag| akash| 2555455|
|Salman| akash| 2555455|
+------+------+--------+

+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|sunny|pratik|  451232|
+-----+------+--------+

# na

x = sqlContext.createDataFrame([(None,"Bob",0.1),("Bob","Carol",None),("Carol",None,0.3),("Bob","Carol",0.2)], ['from','to','amt'])
y = x.na  # returns an object for handling missing values, supports drop, fill, and replace methods
x.show()
print(y)
y.drop().show()

y.fill({'from':'unknown','to':'unknown','amt':0}).show()
y.fill('--').show()
+-----+-----+----+
| from|   to| amt|
+-----+-----+----+
| null|  Bob| 0.1|
|  Bob|Carol|null|
|Carol| null| 0.3|
|  Bob|Carol| 0.2|
+-----+-----+----+

<pyspark.sql.dataframe.DataFrameNaFunctions object at 0x0000021742513940>
+----+-----+---+
|from|   to|amt|
+----+-----+---+
| Bob|Carol|0.2|
+----+-----+---+

+-------+-------+---+
|   from|     to|amt|
+-------+-------+---+
|unknown|    Bob|0.1|
|    Bob|  Carol|0.0|
|  Carol|unknown|0.3|
|    Bob|  Carol|0.2|
+-------+-------+---+

+-----+-----+----+
| from|   to| amt|
+-----+-----+----+
|   --|  Bob| 0.1|
|  Bob|Carol|null|
|Carol|   --| 0.3|
|  Bob|Carol| 0.2|
+-----+-----+----+

# orderBy

x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455)], ['from','to','amt'])

y = x.orderBy(['amt'],ascending=[False])
z = x.orderBy(['amt'],ascending=[True])
x.show()
y.show()
z.show()
+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|sunny|pratik|  451232|
|parag| akash| 2555455|
+-----+------+--------+

+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|parag| akash| 2555455|
|sunny|pratik|  451232|
+-----+------+--------+

+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|sunny|pratik|  451232|
|parag| akash| 2555455|
|vinay|deepak|12466641|
+-----+------+--------+

# PrintSchema

x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455)], ['from','to','amt'])
x.show()
x.printSchema()
+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|sunny|pratik|  451232|
|parag| akash| 2555455|
+-----+------+--------+

root
 |-- from: string (nullable = true)
 |-- to: string (nullable = true)
 |-- amt: long (nullable = true)

# randomSplit

x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455)], ['from','to','amt'])
y = x.randomSplit([0.5,0.5])

x.show()
y[0].show()
y[1].show()


+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|sunny|pratik|  451232|
|parag| akash| 2555455|
+-----+------+--------+

+-----+------+-------+
| from|    to|    amt|
+-----+------+-------+
|sunny|pratik| 451232|
|parag| akash|2555455|
+-----+------+-------+

+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
+-----+------+--------+

# rdd 

x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455)], ['from','to','amt'])
y = x.rdd

x.show()
print(y.collect())
+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|sunny|pratik|  451232|
|parag| akash| 2555455|
+-----+------+--------+

[Row(from='vinay', to='deepak', amt=12466641), Row(from='sunny', to='pratik', amt=451232), Row(from='parag', to='akash', amt=2555455)]

# registerTempTable

x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455)], ['from','to','amt'])
x.registerTempTable(name="TRANS")
y = sqlContext.sql('SELECT * FROM TRANS WHERE amt > 451232')

x.show()
y.show()
+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|sunny|pratik|  451232|
|parag| akash| 2555455|
+-----+------+--------+

+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|parag| akash| 2555455|
+-----+------+--------+

# repartiton 

x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455)], ['from','to','amt'])
y = x.repartition(3)

print(x.rdd.getNumPartitions())
print(y.rdd.getNumPartitions())
y.show()
4
3
+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|parag| akash| 2555455|
|vinay|deepak|12466641|
|sunny|pratik|  451232|
+-----+------+--------+

# replace

x = sqlContext.createDataFrame([("vinay","deepak",12466641),("sunny","pratik",451232),("parag","akash",2555455)], ['from','to','amt'])
y = x.replace('vinay','sunny',['from','to'])

x.show()
y.show()

+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|vinay|deepak|12466641|
|sunny|pratik|  451232|
|parag| akash| 2555455|
+-----+------+--------+

+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|sunny|deepak|12466641|
|sunny|pratik|  451232|
|parag| akash| 2555455|
+-----+------+--------+

# replace

x = sqlContext.createDataFrame([('Sunny',"chirag",0.1),("deepak","vinay",0.2),("Carol","Dave",0.3)], ['from','to','amt'])
y = x.replace('Sunny','Pranav',['from','to'])

x.show()
y.show()
+------+------+---+
|  from|    to|amt|
+------+------+---+
| Sunny|chirag|0.1|
|deepak| vinay|0.2|
| Carol|  Dave|0.3|
+------+------+---+

+------+------+---+
|  from|    to|amt|
+------+------+---+
|Pranav|chirag|0.1|
|deepak| vinay|0.2|
| Carol|  Dave|0.3|
+------+------+---+

#rollup

x = sqlContext.createDataFrame([("vinay","deepak",1246.6641),("sunny","pratik",4512.32),("parag","akash",2555.455)], ['from','to','amt'])
y = x.rollup(['from','to'])
x.show()

print(y)
#y is a grouped data object 
#aggregations will be applied to all numerical columns

y.sum().show()
y.max().show()
y.min().show()
+-----+------+---------+
| from|    to|      amt|
+-----+------+---------+
|vinay|deepak|1246.6641|
|sunny|pratik|  4512.32|
|parag| akash| 2555.455|
+-----+------+---------+

<pyspark.sql.group.GroupedData object at 0x000002C3E1627A58>
+-----+------+---------+
| from|    to| sum(amt)|
+-----+------+---------+
| null|  null|8314.4391|
|vinay|deepak|1246.6641|
|vinay|  null|1246.6641|
|parag| akash| 2555.455|
|parag|  null| 2555.455|
|sunny|  null|  4512.32|
|sunny|pratik|  4512.32|
+-----+------+---------+

+-----+------+---------+
| from|    to| max(amt)|
+-----+------+---------+
| null|  null|  4512.32|
|vinay|deepak|1246.6641|
|vinay|  null|1246.6641|
|parag| akash| 2555.455|
|parag|  null| 2555.455|
|sunny|  null|  4512.32|
|sunny|pratik|  4512.32|
+-----+------+---------+

+-----+------+---------+
| from|    to| min(amt)|
+-----+------+---------+
| null|  null|1246.6641|
|vinay|deepak|1246.6641|
|vinay|  null|1246.6641|
|parag| akash| 2555.455|
|parag|  null| 2555.455|
|sunny|  null|  4512.32|
|sunny|pratik|  4512.32|
+-----+------+---------+

# sample:-
# Returns a stratified sample without replacement based 
# on the fraction given on each stratum.

x = sqlContext.createDataFrame([("vinay","deepak",1246.6641),("sunny","pratik",4512.32),("parag","akash",2555.455)], ['from','to','amt'])
y = x.sample(False,0.5)

x.show()
y.show()
+-----+------+---------+
| from|    to|      amt|
+-----+------+---------+
|vinay|deepak|1246.6641|
|sunny|pratik|  4512.32|
|parag| akash| 2555.455|
+-----+------+---------+

+-----+------+--------+
| from|    to|     amt|
+-----+------+--------+
|sunny|pratik| 4512.32|
|parag| akash|2555.455|
+-----+------+--------+

#schema 
x = sqlContext.createDataFrame([("vinay","deepak",1246.6641),("sunny","pratik",4512.32),("parag","akash",2555.455)], ['from','to','amt'])
y = x.schema
x.show()
print(y)
+-----+------+---------+
| from|    to|      amt|
+-----+------+---------+
|vinay|deepak|1246.6641|
|sunny|pratik|  4512.32|
|parag| akash| 2555.455|
+-----+------+---------+

StructType(List(StructField(from,StringType,true),StructField(to,StringType,true),StructField(amt,DoubleType,true)))

# SlectExpr
x = sqlContext.createDataFrame([("vinay","deepak",1246.6641),("sunny","pratik",4512.32),("parag","akash",2555.455)], ['from','to','amt'])
y = x.selectExpr(['substr(from,1,1)','amt+1000'])

x.show()
y.show()
+-----+------+---------+
| from|    to|      amt|
+-----+------+---------+
|vinay|deepak|1246.6641|
|sunny|pratik|  4512.32|
|parag| akash| 2555.455|
+-----+------+---------+

+---------------------+------------+
|substring(from, 1, 1)|(amt + 1000)|
+---------------------+------------+
|                    v|   2246.6641|
|                    s|     5512.32|
|                    p|    3555.455|
+---------------------+------------+

# show

x = sqlContext.createDataFrame([("vinay","deepak",1246.6641),("sunny","pratik",4512.32),("parag","akash",2555.455)], ['from','to','amt'])
x.show()
+-----+------+---------+
| from|    to|      amt|
+-----+------+---------+
|vinay|deepak|1246.6641|
|sunny|pratik|  4512.32|
|parag| akash| 2555.455|
+-----+------+---------+

# sort

x = sqlContext.createDataFrame([("vinay","deepak",1246.6641),("sunny","pratik",4512.32),("parag","akash",2555.455)], ['from','to','amt'])
y = x.sort(['amt'])

x.show()
y.show()
+-----+------+---------+
| from|    to|      amt|
+-----+------+---------+
|vinay|deepak|1246.6641|
|sunny|pratik|  4512.32|
|parag| akash| 2555.455|
+-----+------+---------+

+-----+------+---------+
| from|    to|      amt|
+-----+------+---------+
|vinay|deepak|1246.6641|
|parag| akash| 2555.455|
|sunny|pratik|  4512.32|
+-----+------+---------+

# sortWithinPartitions
x = sqlContext.createDataFrame([('vinay',"Bobby",0.1,1),("Bobby","sunny",0.2,2),("deepak","parag",0.3,2)], \
                               ['from','to','amt','p_id']).repartition(2,'p_id')
y = x.sortWithinPartitions(['to'])
x.show()
y.show()
print(x.rdd.glom().collect()) # glom() flattens elements on the same partition
print("\n")
print(y.rdd.glom().collect())
+------+-----+---+----+
|  from|   to|amt|p_id|
+------+-----+---+----+
| Bobby|sunny|0.2|   2|
|deepak|parag|0.3|   2|
| vinay|Bobby|0.1|   1|
+------+-----+---+----+

+------+-----+---+----+
|  from|   to|amt|p_id|
+------+-----+---+----+
|deepak|parag|0.3|   2|
| Bobby|sunny|0.2|   2|
| vinay|Bobby|0.1|   1|
+------+-----+---+----+

[[Row(from='Bobby', to='sunny', amt=0.2, p_id=2), Row(from='deepak', to='parag', amt=0.3, p_id=2)], [Row(from='vinay', to='Bobby', amt=0.1, p_id=1)]]


[[Row(from='deepak', to='parag', amt=0.3, p_id=2), Row(from='Bobby', to='sunny', amt=0.2, p_id=2)], [Row(from='vinay', to='Bobby', amt=0.1, p_id=1)]]

# Stat :-Returns a 
# DataFrameStatFunctions for statistic functions.

x = sqlContext.createDataFrame([("vinay","Bobby",0.1,0.001),("Bobby","sunny",0.2,0.02),("sunny","pranav",0.3,0.02)], ['from','to','amt','fees'])
y = x.stat
x.show()
print(y)
print(y.corr(col1="amt",col2="fees"))
+-----+------+---+-----+
| from|    to|amt| fees|
+-----+------+---+-----+
|vinay| Bobby|0.1|0.001|
|Bobby| sunny|0.2| 0.02|
|sunny|pranav|0.3| 0.02|
+-----+------+---+-----+

<pyspark.sql.dataframe.DataFrameStatFunctions object at 0x00000241596F37F0>
0.8660254037844386

# subtract

x = sqlContext.createDataFrame([("vinay","Bobby",0.1,0.001),("Bobby","sunny",0.2,0.02),("sunny","pranav",0.3,0.02)], ['from','to','amt','fees'])
y = sqlContext.createDataFrame([("vinay","Bobby",0.1,0.001),("Bobby","sunny",0.2,0.02),("sunny","pranav",0.3,0.01)], ['from','to','amt','fees'])

z = x.subtract(y)
x.show()
y.show()
z.show()
+-----+------+---+-----+
| from|    to|amt| fees|
+-----+------+---+-----+
|vinay| Bobby|0.1|0.001|
|Bobby| sunny|0.2| 0.02|
|sunny|pranav|0.3| 0.02|
+-----+------+---+-----+

+-----+------+---+-----+
| from|    to|amt| fees|
+-----+------+---+-----+
|vinay| Bobby|0.1|0.001|
|Bobby| sunny|0.2| 0.02|
|sunny|pranav|0.3| 0.01|
+-----+------+---+-----+

+-----+------+---+----+
| from|    to|amt|fees|
+-----+------+---+----+
|sunny|pranav|0.3|0.02|
+-----+------+---+----+

x = sqlContext.createDataFrame([("vinay","Bobby",0.1,0.001),("Bobby","sunny",0.2,0.02),("sunny","pranav",0.3,0.02)], ['from','to','amt','fees'])

y = x.take(num=2)
x.show()
print(y)
+-----+------+---+-----+
| from|    to|amt| fees|
+-----+------+---+-----+
|vinay| Bobby|0.1|0.001|
|Bobby| sunny|0.2| 0.02|
|sunny|pranav|0.3| 0.02|
+-----+------+---+-----+

[Row(from='vinay', to='Bobby', amt=0.1, fees=0.001), Row(from='Bobby', to='sunny', amt=0.2, fees=0.02)]

Conversions

#toDF

x = sqlContext.createDataFrame([('Alice',"Bob",0.1),("Bob","Carol",0.2),("Carol","Dave",0.3)], ['from','to','amt'])
y = x.toDF("seller","buyer","amt")
x.show()
y.show()
+-----+-----+---+
| from|   to|amt|
+-----+-----+---+
|Alice|  Bob|0.1|
|  Bob|Carol|0.2|
|Carol| Dave|0.3|
+-----+-----+---+

+------+-----+---+
|seller|buyer|amt|
+------+-----+---+
| Alice|  Bob|0.1|
|   Bob|Carol|0.2|
| Carol| Dave|0.3|
+------+-----+---+

# toJson
x = sqlContext.createDataFrame([('Alice',"Bob",0.1),("Bob","Carol",0.2),("Carol","Alice",0.3)], ['from','to','amt'])
y = x.toJSON()

x.show()
print(y)
print("\n")
print(y.collect())
+-----+-----+---+
| from|   to|amt|
+-----+-----+---+
|Alice|  Bob|0.1|
|  Bob|Carol|0.2|
|Carol|Alice|0.3|
+-----+-----+---+

MapPartitionsRDD[193] at toJavaRDD at NativeMethodAccessorImpl.java:0


['{"from":"Alice","to":"Bob","amt":0.1}', '{"from":"Bob","to":"Carol","amt":0.2}', '{"from":"Carol","to":"Alice","amt":0.3}']

# toPandas

x = sqlContext.createDataFrame([('Alice',"Bob",0.1),("Bob","Carol",0.2),("Carol","Alice",0.3)], ['from','to','amt'])
y = x.toPandas
x.show()
print(type(y))
y
+-----+-----+---+
| from|   to|amt|
+-----+-----+---+
|Alice|  Bob|0.1|
|  Bob|Carol|0.2|
|Carol|Alice|0.3|
+-----+-----+---+

<class 'method'>





<bound method DataFrame.toPandas of DataFrame[from: string, to: string, amt: double]>

# unionAll

x = sqlContext.createDataFrame([('Alice',"Bob",0.1),("Bob","Carol",0.2),("Carol","Alice",0.3)], ['from','to','amt'])
y = sqlContext.createDataFrame([('sunny',"Bob",0.1),("vinay","Carol",0.2),("pranav","Alice",0.3)], ['from','to','amt'])

z = x.unionAll(y)

x.show()
y.show()
z.show()
+-----+-----+---+
| from|   to|amt|
+-----+-----+---+
|Alice|  Bob|0.1|
|  Bob|Carol|0.2|
|Carol|Alice|0.3|
+-----+-----+---+

+------+-----+---+
|  from|   to|amt|
+------+-----+---+
| sunny|  Bob|0.1|
| vinay|Carol|0.2|
|pranav|Alice|0.3|
+------+-----+---+

+------+-----+---+
|  from|   to|amt|
+------+-----+---+
| Alice|  Bob|0.1|
|   Bob|Carol|0.2|
| Carol|Alice|0.3|
| sunny|  Bob|0.1|
| vinay|Carol|0.2|
|pranav|Alice|0.3|
+------+-----+---+

# unpersist

x = sqlContext.createDataFrame([('Alice',"Bob",0.1),("Bob","Carol",0.2),("Carol","Alice",0.3)], ['from','to','amt'])
x.cache()
x.count()
x.show()

print(x.is_cached)
x.unpersist()
print(x.is_cached)
+-----+-----+---+
| from|   to|amt|
+-----+-----+---+
|Alice|  Bob|0.1|
|  Bob|Carol|0.2|
|Carol|Alice|0.3|
+-----+-----+---+

True
False

# where 
x = sqlContext.createDataFrame([('Alice',"Bob",0.1),("Bob","Carol",0.2),("Carol","Alice",0.3)], ['from','to','amt'])
y = x.where("amt > 0.2")

x.show()
y.show()
+-----+-----+---+
| from|   to|amt|
+-----+-----+---+
|Alice|  Bob|0.1|
|  Bob|Carol|0.2|
|Carol|Alice|0.3|
+-----+-----+---+

+-----+-----+---+
| from|   to|amt|
+-----+-----+---+
|Carol|Alice|0.3|
+-----+-----+---+

# withColumn

x = sqlContext.createDataFrame([('Alice',"Bob",0.1),("Bob","Carol",0.2),("Carol","Alice",0.3)], ['from','to','amt'])
y = x.withColumn('conf',x.amt.isNotNull())

x.show()
y.show()
+-----+-----+---+
| from|   to|amt|
+-----+-----+---+
|Alice|  Bob|0.1|
|  Bob|Carol|0.2|
|Carol|Alice|0.3|
+-----+-----+---+

+-----+-----+---+----+
| from|   to|amt|conf|
+-----+-----+---+----+
|Alice|  Bob|0.1|true|
|  Bob|Carol|0.2|true|
|Carol|Alice|0.3|true|
+-----+-----+---+----+

# withColumnRenamed
x = sqlContext.createDataFrame([('Alice',"Bob",0.1),("Bob","Carol",0.2),("Carol","Dave",0.3)], ['from','to','amt'])
y = x.withColumnRenamed('amt','amount')
x.show()
y.show()
+-----+-----+---+
| from|   to|amt|
+-----+-----+---+
|Alice|  Bob|0.1|
|  Bob|Carol|0.2|
|Carol| Dave|0.3|
+-----+-----+---+

+-----+-----+------+
| from|   to|amount|
+-----+-----+------+
|Alice|  Bob|   0.1|
|  Bob|Carol|   0.2|
|Carol| Dave|   0.3|
+-----+-----+------+

# write
import json
x = sqlContext.createDataFrame([('Alice',"Bob",0.1),("Bob","Carol",0.2),("Carol","Dave",0.3)], ['from','to','amt'])
y = x.write.mode('overwrite').json('./dataframeWriteExample.json')
x.show()


# Read the DF back in from file
sqlContext.read.json('./dataframeWriteExample.json').show()
+-----+-----+---+
| from|   to|amt|
+-----+-----+---+
|Alice|  Bob|0.1|
|  Bob|Carol|0.2|
|Carol| Dave|0.3|
+-----+-----+---+

+---+-----+-----+
|amt| from|   to|
+---+-----+-----+
|0.3|Carol| Dave|
|0.1|Alice|  Bob|
|0.2|  Bob|Carol|
+---+-----+-----+



   



      


  
      

      

        

          

            

  
About


      

        pyspark dataframe made easy 
      


    
Topics

    

        

      
  python

      
  api

      
  json

      
  csv

      
  spark

      
  filter

      
  bigdata

      
  apache

      
  pandas

      
  pyspark

      
  join

      
  parquet

      
  dataframe

      
  databricks

      
  rdd

      
  groupby

      
  agg

      
  coalesce

      
  cca175

      
  bigqu

  


    


    
Resources

    

      
        
        Readme
    


  



  
  

  

    
      Activity  



  
Stars

  

    
      16
      stars  


  
Watchers

  

    
      2
      watching  


  
Forks

  

    
      5
      forks  


    

      
          Report repository
    




          

        


        
            

              

                

  Releases


    
No releases published


              

            


        
        
            

              

                
  

  Packages
      



      

        No packages published 

      




              

            


        
            

        
        
        
            

              

                
Languages