Import csv file contents into pyspark dataframes

Question

How can I import a .csv file into pyspark dataframes? I even tried to read csv file in Pandas and then convert it to a spark dataframe using createDataFrame, but it is still showing some error. Can someone guide me through this? Also, please tell me how can I import an xlsx file? I'm trying to import csv content into pandas dataframes and then convert it into spark data frames, but it is showing the error:

"Py4JJavaError" An error occurred while calling o28.applySchemaToPythonRDD. : java.lang.RuntimeException: java.lang.RuntimeException: Unable to instantiate org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClient 

My code is:

from pyspark import SparkContext 
from pyspark.sql import SQLContext 
import pandas as pd 
sqlc=SQLContext(sc) 
df=pd.read_csv(r'D:\BestBuy\train.csv') 
sdf=sqlc.createDataFrame(df) 

Answer 1

"How can I import a .csv file into pyspark dataframes ?" -- there are many ways to do this; the simplest would be to start up pyspark with Databrick's spark-csv module. You can do this by starting pyspark with

pyspark --packages com.databricks:spark-csv_2.10:1.4.0

then you can follow the following steps:

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

df = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('cars.csv')

The other method would be to read in the text file as an rdd using

myrdd = sc.textFile("yourfile.csv").map(lambda line: line.split(","))

Then transform your data so that every item is in the correct format for the schema (i.e. Ints, Strings, Floats, etc.). You'll want to then use

>>> from pyspark.sql import Row
>>> Person = Row('name', 'age')
>>> person = rdd.map(lambda r: Person(*r))
>>> df2 = sqlContext.createDataFrame(person)
>>> df2.collect()
[Row(name=u'Alice', age=1)]
>>> from pyspark.sql.types import *
>>> schema = StructType([
...    StructField("name", StringType(), True),
...    StructField("age", IntegerType(), True)])
>>> df3 = sqlContext.createDataFrame(rdd, schema)
>>> df3.collect()
[Row(name=u'Alice', age=1)]

Reference: http://spark.apache.org/docs/1.6.1/api/python/pyspark.sql.html#pyspark.sql.Row

"Also, please tell me how can I import xlsx file?" -- Excel files are not used in "Big Data"; Spark is meant to be used with large files or databases. If you have an Excel file that is 50GB in size, then you're doing things wrong. Excel wouldn't even be able to open a file that size; from my experience, anything above 20MB and Excel dies.

Answer 2

Following worked well for me:

from pyspark.sql.types import *
schema = StructType([StructField("name", StringType(), True),StructField("age", StringType(), True)]
pd_df = pd.read_csv("<inputcsvfile>")
sp_df = spark.createDataFrame(pd_df, schema=schema)

Answer 3

I have in my local directory a file 'temp.csv'. From there, using a local instance I do the following:

>>> from pyspark import SQLContext
>>> from pyspark.sql import Row
>>> sql_c = SQLContext(sc)
>>> d0 = sc.textFile('./temp.csv')
>>> d0.collect()
[u'a,1,.2390', u'b,2,.4390', u'c,3,.2323']
>>> d1 = d0.map(lambda x: x.split(',')).map(lambda x: Row(label = x[0], number = int(x[1]), value = float(x[2])))
>>> d1.take(1)
[Row(label=u'a', number=1, value=0.239)]
>>> df = sql_c.createDataFrame(d1)
>>> df_cut = df[df.number>1]
>>> df_cut.select('label', 'value').collect()
[Row(label=u'b', value=0.439), Row(label=u'c', value=0.2323)]

So d0 is the raw text file that we send off to a spark RDD. In order for you to make a data frame, you want to break the csv apart, and to make every entry a Row type, as I do when creating d1. The last step is to make the data frame from the RDD.

Answer 4

You can use the package spark-csv by DataBricks that does a lot of things for you automatically, like taking care of the header, use escape characters, automatic schema inferring etcetera. Starting from Spark 2.0 there is an inbuilt function for dealing with CSVs.