RDDs and DataFrames

  • Creating RDDs and DataFrames using SparkContext
  • Interoperability between RDDs and DataFrames
  • Multiple rows and multiple column specifications for DataFrames
  • Creating DataFrames using SQLContext
  • Selecting, editing and renaming columns in dataframes
  • Interoperability between Pandas and Spark dataframes
In [1]:
import findspark
findspark.init()
In [3]:
# import pyspark # only run after findspark.init()
# from pyspark.sql import SparkSession
# spark = SparkSession.builder.getOrCreate()

# df = spark.sql('''select 'spark' as hello ''')
# df.show()
In [2]:
import pyspark
from pyspark import SparkContext
sc = SparkContext.getOrCreate()
In [3]:
sc
Out[3]:

SparkContext

Spark UI

Version
v2.4.1
Master
local[*]
AppName
pyspark-shell
In [4]:
from pyspark.sql.types import Row
from datetime import datetime

Creating RDDs using sc.parallelize()

In [31]:
simple_data = sc.parallelize([1, "Alice", 50])
simple_data
Out[31]:
ParallelCollectionRDD[12] at parallelize at PythonRDD.scala:195
In [7]:
simple_data.count()
Out[7]:
3
In [8]:
simple_data.first()
Out[8]:
1
In [9]:
simple_data.take(2)
Out[9]:
[1, 'Alice']
In [10]:
simple_data.collect()
Out[10]:
[1, 'Alice', 50]

This is an ERROR!

  • This RDD does not have "columns", it cannot be represented as a tabular data frame
  • DataFrames are structured datasets
In [ ]:
df = simple_data.toDF()

RDDs with records using sc.parallelize()

In [32]:
records = sc.parallelize([[1, "Alice", 50], [2, "Bob", 80]])
records
Out[32]:
ParallelCollectionRDD[13] at parallelize at PythonRDD.scala:195
In [13]:
records.collect()
Out[13]:
[[1, 'Alice', 50], [2, 'Bob', 80]]
In [14]:
records.count()
Out[14]:
2
In [15]:
records.first()
Out[15]:
[1, 'Alice', 50]
In [16]:
records.take(2)
Out[16]:
[[1, 'Alice', 50], [2, 'Bob', 80]]
In [17]:
records.collect()
Out[17]:
[[1, 'Alice', 50], [2, 'Bob', 80]]

This is an NOT an error!

  • This RDD does have "columns", it can be represented as a tabular data frame
In [33]:
df = records.toDF()
In [34]:
df
Out[34]:
DataFrame[_1: bigint, _2: string, _3: bigint]
In [50]:
df.show()
+---+-----+---+
| _1|   _2| _3|
+---+-----+---+
|  1|Alice| 50|
|  2|  Bob| 80|
+---+-----+---+

Creating dataframes using sc.parallelize() and Row() functions

  • Row functions allow specifying column names for dataframes
In [35]:
data = sc.parallelize([Row(id=1,
                           name="Alice",
                           score=50)])
data
Out[35]:
ParallelCollectionRDD[20] at parallelize at PythonRDD.scala:195
In [36]:
data.count()
Out[36]:
1
In [37]:
data.collect()
Out[37]:
[Row(id=1, name='Alice', score=50)]
In [38]:
df = data.toDF()
df.show()
+---+-----+-----+
| id| name|score|
+---+-----+-----+
|  1|Alice|   50|
+---+-----+-----+

Working with multiple rows

In [39]:
data = sc.parallelize([Row(
                           id=1,
                           name="Alice",
                           score=50
                        ),
                        Row(
                            id=2,
                            name="Bob",
                            score=80
                        ),
                        Row(
                            id=3,
                            name="Charlee",
                            score=75
                        )])
In [40]:
df = data.toDF()
df.show()
+---+-------+-----+
| id|   name|score|
+---+-------+-----+
|  1|  Alice|   50|
|  2|    Bob|   80|
|  3|Charlee|   75|
+---+-------+-----+

Multiple columns with complex data types

In [41]:
complex_data = sc.parallelize([Row(
                                col_float=1.44,
                                col_integer=10,
                                col_string="John")
                           ])
In [42]:
complex_data_df = complex_data.toDF()
complex_data_df.show()
+---------+-----------+----------+
|col_float|col_integer|col_string|
+---------+-----------+----------+
|     1.44|         10|      John|
+---------+-----------+----------+

In [43]:
complex_data = sc.parallelize([Row(
                                col_float=1.44, 
                                col_integer=10, 
                                col_string="John", 
                                col_boolean=True, 
                                col_list=[1, 2, 3])
                           ])
In [44]:
complex_data_df = complex_data.toDF()
complex_data_df.show()
+-----------+---------+-----------+---------+----------+
|col_boolean|col_float|col_integer| col_list|col_string|
+-----------+---------+-----------+---------+----------+
|       true|     1.44|         10|[1, 2, 3]|      John|
+-----------+---------+-----------+---------+----------+

In [45]:
complex_data = sc.parallelize([Row(
                                col_list = [1, 2, 3], 
                                col_dict = {"k1": 0, "k2": 1, "k3": 2}, 
                                col_row = Row(columnA = 10, columnB = 20, columnC = 30), 
                                col_time = datetime(2014, 8, 1, 14, 1, 5)
                            )])
In [46]:
complex_data_df = complex_data.toDF()
complex_data_df.show()
+--------------------+---------+------------+-------------------+
|            col_dict| col_list|     col_row|           col_time|
+--------------------+---------+------------+-------------------+
|[k3 -> 2, k1 -> 0...|[1, 2, 3]|[10, 20, 30]|2014-08-01 14:01:05|
+--------------------+---------+------------+-------------------+

Multiple rows with complex data types

In [47]:
complex_data = sc.parallelize([Row(
                                col_list = [1, 2, 3],
                                col_dict = {"k1": 0},
                                col_row = Row(a=10, b=20, c=30),
                                col_time = datetime(2014, 8, 1, 14, 1, 5)
                            ),              
                            Row(
                                col_list = [1, 2, 3, 4, 5], 
                                col_dict = {"k1": 0,"k2": 1 }, 
                                col_row = Row(a=40, b=50, c=60),
                                col_time = datetime(2014, 8, 2, 14, 1, 6)
                            ),
                            Row(
                                col_list = [1, 2, 3, 4, 5, 6, 7], 
                                col_dict = {"k1": 0, "k2": 1, "k3": 2 }, 
                                col_row = Row(a=70, b=80, c=90),
                                col_time = datetime(2014, 8, 3, 14, 1, 7)
                            )]) 
In [48]:
complex_data_df = complex_data.toDF()
complex_data_df.show()
+--------------------+--------------------+------------+-------------------+
|            col_dict|            col_list|     col_row|           col_time|
+--------------------+--------------------+------------+-------------------+
|           [k1 -> 0]|           [1, 2, 3]|[10, 20, 30]|2014-08-01 14:01:05|
|  [k1 -> 0, k2 -> 1]|     [1, 2, 3, 4, 5]|[40, 50, 60]|2014-08-02 14:01:06|
|[k3 -> 2, k1 -> 0...|[1, 2, 3, 4, 5, 6...|[70, 80, 90]|2014-08-03 14:01:07|
+--------------------+--------------------+------------+-------------------+

Creating DataFrames using SQLContext

  • SQLContext can create dataframes directly from raw data
In [9]:
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
In [10]:
sqlContext
Out[10]:
<pyspark.sql.context.SQLContext at 0x18eab17e860>
In [52]:
df = sqlContext.range(5)
df
Out[52]:
DataFrame[id: bigint]
In [100]:
df.show()
+---+
| id|
+---+
|  0|
|  1|
|  2|
|  3|
|  4|
+---+

In [53]:
df.count()
Out[53]:
5

Rows specified in tuples

In [54]:
data = [('Alice', 50),
        ('Bob', 80),
        ('Charlee', 75)]
In [55]:
sqlContext.createDataFrame(data).show()
+-------+---+
|     _1| _2|
+-------+---+
|  Alice| 50|
|    Bob| 80|
|Charlee| 75|
+-------+---+

In [56]:
sqlContext.createDataFrame(data, ['Name', 'Score']).show()
+-------+-----+
|   Name|Score|
+-------+-----+
|  Alice|   50|
|    Bob|   80|
|Charlee|   75|
+-------+-----+

In [12]:
complex_data = [
                 (1.0,
                  10,
                  "Alice", 
                  True, 
                  [1, 2, 3], 
                  {"k1": 0},
                  Row(a=1, b=2, c=3), 
                  datetime(2014, 8, 1, 14, 1, 5)),

                 (2.0,
                  20,
                  "Bob", 
                  True, 
                  [1, 2, 3, 4, 5], 
                  {"k1": 0,"k2": 1 }, 
                  Row(a=1, b=2, c=3), 
                  datetime(2014, 8, 1, 14, 1, 5)),

                  (3.0,
                   30,
                   "Charlee", 
                   False, 
                   [1, 2, 3, 4, 5, 6], 
                   {"k1": 0, "k2": 1, "k3": 2 }, 
                   Row(a=1, b=2, c=3), 
                   datetime(2014, 8, 1, 14, 1, 5))
                ] 
In [13]:
sqlContext.createDataFrame(complex_data).show()
+---+---+-------+-----+------------------+--------------------+---------+-------------------+
| _1| _2|     _3|   _4|                _5|                  _6|       _7|                 _8|
+---+---+-------+-----+------------------+--------------------+---------+-------------------+
|1.0| 10|  Alice| true|         [1, 2, 3]|           [k1 -> 0]|[1, 2, 3]|2014-08-01 14:01:05|
|2.0| 20|    Bob| true|   [1, 2, 3, 4, 5]|  [k1 -> 0, k2 -> 1]|[1, 2, 3]|2014-08-01 14:01:05|
|3.0| 30|Charlee|false|[1, 2, 3, 4, 5, 6]|[k3 -> 2, k1 -> 0...|[1, 2, 3]|2014-08-01 14:01:05|
+---+---+-------+-----+------------------+--------------------+---------+-------------------+

In [14]:
complex_data_df = sqlContext.createDataFrame(complex_data, [
        'col_integer',
        'col_float',
        'col_string',
        'col_boolean',
        'col_list',
        'col_dictionary',
        'col_row',
        'col_date_time']
    )
complex_data_df.show()
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|col_integer|col_float|col_string|col_boolean|          col_list|      col_dictionary|  col_row|      col_date_time|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|        1.0|       10|     Alice|       true|         [1, 2, 3]|           [k1 -> 0]|[1, 2, 3]|2014-08-01 14:01:05|
|        2.0|       20|       Bob|       true|   [1, 2, 3, 4, 5]|  [k1 -> 0, k2 -> 1]|[1, 2, 3]|2014-08-01 14:01:05|
|        3.0|       30|   Charlee|      false|[1, 2, 3, 4, 5, 6]|[k3 -> 2, k1 -> 0...|[1, 2, 3]|2014-08-01 14:01:05|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+

Creating dataframes using SQL Context and the Row function

  • Row functions can be used without specifying column names
In [60]:
data = sc.parallelize([
    Row(1, "Alice", 50),
    Row(2, "Bob", 80),
    Row(3, "Charlee", 75)
])
In [61]:
column_names = Row('id', 'name', 'score')  
students = data.map(lambda r: column_names(*r))
In [62]:
students
Out[62]:
PythonRDD[131] at RDD at PythonRDD.scala:53
In [63]:
students.collect()
Out[63]:
[Row(id=1, name='Alice', score=50),
 Row(id=2, name='Bob', score=80),
 Row(id=3, name='Charlee', score=75)]
In [64]:
students_df = sqlContext.createDataFrame(students)
students_df
Out[64]:
DataFrame[id: bigint, name: string, score: bigint]
In [65]:
students_df.show()
+---+-------+-----+
| id|   name|score|
+---+-------+-----+
|  1|  Alice|   50|
|  2|    Bob|   80|
|  3|Charlee|   75|
+---+-------+-----+

Extracting specific rows from dataframes

In [66]:
complex_data_df.first()
Out[66]:
Row(col_integer=1.0, col_float=10, col_string='Alice', col_boolean=True, col_list=[1, 2, 3], col_dictionary={'k1': 0}, col_row=Row(a=1, b=2, c=3), col_date_time=datetime.datetime(2014, 8, 1, 14, 1, 5))
In [67]:
complex_data_df.take(2)
Out[67]:
[Row(col_integer=1.0, col_float=10, col_string='Alice', col_boolean=True, col_list=[1, 2, 3], col_dictionary={'k1': 0}, col_row=Row(a=1, b=2, c=3), col_date_time=datetime.datetime(2014, 8, 1, 14, 1, 5)),
 Row(col_integer=2.0, col_float=20, col_string='Bob', col_boolean=True, col_list=[1, 2, 3, 4, 5], col_dictionary={'k1': 0, 'k2': 1}, col_row=Row(a=1, b=2, c=3), col_date_time=datetime.datetime(2014, 8, 1, 14, 1, 5))]

Extracting specific cells from dataframes

In [68]:
cell_string = complex_data_df.collect()[0][2]
cell_string
Out[68]:
'Alice'
In [69]:
cell_list = complex_data_df.collect()[0][4]
cell_list
Out[69]:
[1, 2, 3]
In [70]:
cell_list.append(100)
cell_list
Out[70]:
[1, 2, 3, 100]
In [71]:
complex_data_df.show()
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|col_integer|col_float|col_string|col_boolean|          col_list|      col_dictionary|  col_row|      col_date_time|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|        1.0|       10|     Alice|       true|         [1, 2, 3]|           [k1 -> 0]|[1, 2, 3]|2014-08-01 14:01:05|
|        2.0|       20|       Bob|       true|   [1, 2, 3, 4, 5]|  [k1 -> 0, k2 -> 1]|[1, 2, 3]|2014-08-01 14:01:05|
|        3.0|       30|   Charlee|      false|[1, 2, 3, 4, 5, 6]|[k3 -> 2, k1 -> 0...|[1, 2, 3]|2014-08-01 14:01:05|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+

Selecting specific columns

In [15]:
complex_data_df.rdd\
    .map(lambda x: (x.col_string, x.col_dictionary))\
    .collect()
Out[15]:
[('Alice', {'k1': 0}),
 ('Bob', {'k1': 0, 'k2': 1}),
 ('Charlee', {'k3': 2, 'k1': 0, 'k2': 1})]
In [16]:
complex_data_df.select(
    'col_string',
    'col_list',
    'col_date_time'
).show()
+----------+------------------+-------------------+
|col_string|          col_list|      col_date_time|
+----------+------------------+-------------------+
|     Alice|         [1, 2, 3]|2014-08-01 14:01:05|
|       Bob|   [1, 2, 3, 4, 5]|2014-08-01 14:01:05|
|   Charlee|[1, 2, 3, 4, 5, 6]|2014-08-01 14:01:05|
+----------+------------------+-------------------+

Editing columns

In [17]:
complex_data_df.rdd\
           .map(lambda x: (x.col_string + " Boo"))\
           .collect()
Out[17]:
['Alice Boo', 'Bob Boo', 'Charlee Boo']

Adding a column

In [18]:
complex_data_df.select(
                   'col_integer',
                   'col_float'
            )\
           .withColumn(
                   "col_sum",
                    complex_data_df.col_integer + complex_data_df.col_float
           )\
           .show()
+-----------+---------+-------+
|col_integer|col_float|col_sum|
+-----------+---------+-------+
|        1.0|       10|   11.0|
|        2.0|       20|   22.0|
|        3.0|       30|   33.0|
+-----------+---------+-------+

In [19]:
complex_data_df.select('col_boolean')\
               .withColumn(
                   "col_opposite",
                   complex_data_df.col_boolean == False )\
               .show()
+-----------+------------+
|col_boolean|col_opposite|
+-----------+------------+
|       true|       false|
|       true|       false|
|      false|        true|
+-----------+------------+

Editing a column name

In [225]:
complex_data_df.withColumnRenamed("col_dictionary","col_map").show()
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|col_integer|col_float|col_string|col_boolean|          col_list|             col_map|  col_row|      col_date_time|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|        1.0|       10|     Alice|       true|         [1, 2, 3]|           [k1 -> 0]|[1, 2, 3]|2014-08-01 14:01:05|
|        2.0|       20|       Bob|       true|   [1, 2, 3, 4, 5]|  [k1 -> 0, k2 -> 1]|[1, 2, 3]|2014-08-01 14:01:05|
|        3.0|       30|   Charlee|      false|[1, 2, 3, 4, 5, 6]|[k3 -> 2, k1 -> 0...|[1, 2, 3]|2014-08-01 14:01:05|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+

In [226]:
complex_data_df.select(complex_data_df.col_string.alias("Name")).show()
+-------+
|   Name|
+-------+
|  Alice|
|    Bob|
|Charlee|
+-------+

Interoperablity between Pandas dataframe and Spark dataframe

In [20]:
import pandas
In [21]:
df_pandas = complex_data_df.toPandas()
df_pandas
Out[21]:
col_integer col_float col_string col_boolean col_list col_dictionary col_row col_date_time
0 1.0 10 Alice True [1, 2, 3] {'k1': 0} (1, 2, 3) 2014-08-01 14:01:05
1 2.0 20 Bob True [1, 2, 3, 4, 5] {'k1': 0, 'k2': 1} (1, 2, 3) 2014-08-01 14:01:05
2 3.0 30 Charlee False [1, 2, 3, 4, 5, 6] {'k3': 2, 'k1': 0, 'k2': 1} (1, 2, 3) 2014-08-01 14:01:05
In [235]:
df_spark = sqlContext.createDataFrame(df_pandas).show()  
df_spark
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|col_integer|col_float|col_string|col_boolean|          col_list|      col_dictionary|  col_row|      col_date_time|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+
|        1.0|       10|     Alice|       true|         [1, 2, 3]|           [k1 -> 0]|[1, 2, 3]|2014-08-01 14:01:05|
|        2.0|       20|       Bob|       true|   [1, 2, 3, 4, 5]|  [k1 -> 0, k2 -> 1]|[1, 2, 3]|2014-08-01 14:01:05|
|        3.0|       30|   Charlee|      false|[1, 2, 3, 4, 5, 6]|[k3 -> 2, k1 -> 0...|[1, 2, 3]|2014-08-01 14:01:05|
+-----------+---------+----------+-----------+------------------+--------------------+---------+-------------------+

In [ ]: