Zeppelin and Spark: Merge Multiple CSVs into Parquet

Introduction

The purpose of this article is to demonstrate how to load multiple CSV files on an HDFS filesystem into a single Dataframe and write to Parquet.

Two approaches are demonstrated.  The first approach is not recommended, but is shown for completeness.

First Approach

One approach might be to define each path:

%pyspark

import locale
locale.setlocale(locale.LC_ALL, 'en_US')

p1 = "/data/output/followers/mitshu/ec2-52-39-251-219.us-west-2.compute.amazonaws.com/0-ec2-52-39-251-219.us-west-2.compute.amazonaws.com/twitterFollowers.csv"
p2 = "/data/output/followers/mitshu/ec2-52-42-100-207.us-west-2.compute.amazonaws.com/0-ec2-52-42-100-207.us-west-2.compute.amazonaws.com/twitterFollowers.csv"
p3 = "/data/output/followers/mitshu/ec2-52-42-198-4.us-west-2.compute.amazonaws.com/0-ec2-52-42-198-4.us-west-2.compute.amazonaws.com/twitterFollowers.csv"
p4 = "/data/output/followers/mitshu/ec2-54-70-37-224.us-west-2.compute.amazonaws.com/0-ec2-54-70-37-224.us-west-2.compute.amazonaws.com/twitterFollowers.csv"

and then open each CSV at that path as an RDD and transform to a dataframe:

%pyspark

rdd_m1 = sc.textFile(p1)
print rdd_m1.take(5)

df_m1 = rdd_m1.\
    map(lambda x: x.split("\t")).\
    filter(lambda x: len(x) == 6). \
    map(lambda x: {
        'id':x[0],
        'profile_id':x[1],
        'profile_name':x[2],
        'follower_id':x[3],
        'follwer_name':x[4],
        'unknown':x[5]})\
    .toDF()
df_m1.limit(5).show()
df_m1.registerTempTable("df_m1")

This would need to be repeated for each dataframe.

The dataframes could then be merged using the unionAll operator.

%pyspark
import pandas as pd

df = df_m1.unionAll(df_m2).unionAll(df_m3).unionAll(df_m4)

print "DF 1: {0}".format(df_m1.count())
print "DF 2: {0}".format(df_m2.count())
print "DF 3: {0}".format(df_m3.count())
print "DF 4: {0}".format(df_m4.count())
print "Merged Dataframe: {0}".format(df.count())

and finally written to parquet.

%pyspark

df.write.parquet("/data/output/followers/mitshu/joined.prq")

Easier Approach

Notice the convenient way of reading multiple CSV in nested directories into a single RDD:

%pyspark

path="/data/output/followers/mitshu/*/*/*.csv"
rdd = sc.textFile(path)
print "count = {}".format(rdd.count())

This is clearly better than defining each path individually.


There are multiple ways to transform RDDs into Dataframes (DFs):

%pyspark

def to_json(r):
    j = {}
    t = r.split("\t")
    j['num_followers'] = t[0]
    j['followed_userid'] = t[1]
    j['followed_handle'] = t[2]
    j['follower_userid'] = t[3]
    j['follower_handle'] = t[4]
    return j
    
df = rdd.map(to_json).toDF()
print "count = {}".format(df.count())
df.show()

This is not necessarily superior to the first approach; but it is an alternative to consider.

 

Load from Parquet

For subsequent analysis, load from Parquet using this code:

%pyspark

df = sqlContext.read.parquet("/data/output/followers/mitshu/joined.prq")
df.limit(5).show()

 

References

1. [Blogger] Writing to Parquet

Zeppelin and Spark: Transforming a CSV to Parquet

Transform a CSV file to Parquet Format

Apache Parquet is a columnar storage format available to any project in the Hadoop ecosystem.  Parquet is built to support very efficient compression and encoding schemes.  Twitter is starting to convert some of its major data source to Parquet in order to take advantage of the compression and deserialization savings.

CSV to RDD

Load the CSV file into an RDD

%pyspark

rdd = sc.textFile("/path/to/input/myfile.csv")
print rdd.take(5)

and the output looks like this:

[u'19999174,Logicknot,Cameron MacKinnon,0,Male, , , ', u'433249647,LogieTatjana,Tatjana Logie \u2653,0,Female, , , ', u'1346426538,Logistic_soares,Luciana Soares,0,Female, , , ', u'18355981,Loh69,Laurent Grad,0,Male, , , ', u'2976559335,LohMarta,Marta Loh,0,Female, , , ']

RDD to DF

Now we need to convert the RDD into a Dataframe (DF):

%pyspark

df = rdd.\
    map(lambda x: x.split(",")).\
    filter(lambda x: len(x) == 8). \
    map(lambda x: {
        'userid':x[0],
        'twitterhandle':x[1],
        'full_name':x[2],
        'country':x[3],
        'gender':x[4],
        'age':x[5],
        'age_min':x[6],
        'age_max':x[7]})\
    .toDF()
df.limit(5).show()

The bad news is that this code might look complicated. The good news is that the code is a lot simpler than it appears.

The first map condition splits each record on the delimiter (a comma).  Now we're dealing with a list of 8 tokens per record.  The second condition (the filter) will reject any line that does not have 8 tokens.  The third, and final, map condition will take each token in the list and create a heading for it.

The output looks like this:

+---+-------+-------+-------+-----------------+------+---------------+----------+
|age|age_max|age_min|country|        full_name|gender|  twitterhandle|    userid|
+---+-------+-------+-------+-----------------+------+---------------+----------+
|   |       |       |      0|Cameron MacKinnon|  Male|      Logicknot|  19999174|
|   |       |       |      0|  Tatjana Logie ♓|Female|   LogieTatjana| 433249647|
|   |       |       |      0|   Luciana Soares|Female|Logistic_soares|1346426538|
|   |       |       |      0|     Laurent Grad|  Male|          Loh69|  18355981|
|   |       |       |      0|        Marta Loh|Female|       LohMarta|2976559335|
+---+-------+-------+-------+-----------------+------+---------------+----------+

Note that we could have greatly simplified the code by just doing this:

%pyspark

df = rdd.\
    map(lambda x: x.split(",")).toDF()
df.limit(5).show()

And then we would have got this:

+----------+---------------+-----------------+---+------+---+---+---+
|        _1|             _2|               _3| _4|    _5| _6| _7| _8|
+----------+---------------+-----------------+---+------+---+---+---+
|  19999174|      Logicknot|Cameron MacKinnon|  0|  Male|   |   |   |
| 433249647|   LogieTatjana|  Tatjana Logie ♓|  0|Female|   |   |   |
|1346426538|Logistic_soares|   Luciana Soares|  0|Female|   |   |   |
|  18355981|          Loh69|     Laurent Grad|  0|  Male|   |   |   |
|2976559335|       LohMarta|        Marta Loh|  0|Female|   |   |   |
+----------+---------------+-----------------+---+------+---+---+---+

But presumably it's important to have column headers; hence the additional code in the first snippet above.

DF to Parquet

The final step is to transform the Dataframe into a Parquet file.

This can be accomplished in a single line:

%pyspark

df.write.parquet("/path/to/output/myfile")

Parquet to DF

Read the parquet file using this code:

%pyspark

df = sqlContext.read.parquet("/path/to/output/myfile")
df.limit(5).show()

And the output, as expected, looks like this:

+---+-------+-------+-------+----------------+------+---------------+----------+
|age|age_max|age_min|country|       full_name|gender|  twitterhandle|    userid|
+---+-------+-------+-------+----------------+------+---------------+----------+
|   |       |       |      0|nelson rodriguez|  Male| ojooooopublico|2771575827|
|   |       |       |      0|    Mary Garciaа|Female|  ojor_ozefofuw|3056364751|
|   |       |       |      0|   Andres nucleo|  Male|ojosrastafara12|1035247920|
|   |       |       |      0|          Omaira|Female|      ojovalles| 183602059|
|   |       |       |      0|   Olivia Scotti|Female|       ojscotti|3072660401|
+---+-------+-------+-------+----------------+------+---------------+----------+

This saved us a significant amount of time. By saving as a parquet file, we not only achieve space efficiency on our cluster, but have the ability to rapidly load the parquet data into a Dataframe with the column headings we specified earlier.