How to elegantly create a pyspark Dataframe from a csv file and convert it to a Pandas Dataframe?
<p>I'm having a CSV file which I want to read into an RDD or DataFrame. This is working so far, but if I collect the data and convert it into a pandas DataFrame for plotting the table is "malformed".</p>

<p>Here is how I read the CSV file:</p>

<pre><code>NUMERIC_DATA_FILE = os.path.join(DATA_DIR, "train_numeric.csv")
numeric_rdd = sc.textFile(NUMERIC_DATA_FILE)
numeric_rdd = numeric_rdd.mapPartitions(lambda x: csv.reader(x, delimiter=","))
numeric_df = sqlContext.createDataFrame(numeric_rdd)
numeric_df.registerTempTable("numeric")
</code></pre>

<p>The result looks like this:</p>

<p><a href="https://i.stack.imgur.com/fRwjI.png" rel="nofollow"><img src="https://i.stack.imgur.com/fRwjI.png" alt="enter image description here"></a></p>

<p>Is there an easy way to correctly set the first row of the CSV data to columns and the first column as index?</p>

<hr>

<p>This problem goes further as I try to select data from the <code>DataFrame</code>:</p>

<pre><code>numeric_df.select("SELECT Id FROM numeric")
</code></pre>

<p>which gives me:</p>

<pre><code>AnalysisException: u"cannot resolve 'SELECT Id FROM numeric' given input columns _799, _640, _963, _70, _364, _143, _167, 
_156, _553, _835, _780, _235, ...
</code></pre>

<h2>Setting a schema for your PySpark DataFrame</h2>

<p>Your PySpark DataFrame does not have a schema assigned to it. You should replace your code with the snippet below:</p>

<pre><code>from pyspark.sql.types import *
NUMERIC_DATA_FILE = sc.textFile(os.path.join(DATA_DIR, "train_numeric.csv"))

# Extract the header line
header = NUMERIC_DATA_FILE.first()

# Assuming that all the columns are numeric, let's create a new StructField for each column
fields = [StructField(field_name, FloatType(), True) for field_name in header]
</code></pre>

<p>Now, we can construct our schema,     </p>

<pre><code>schema = StructType(fields)

# We have the remove the header from the textfile rdd

# Extracting the header (first line) from the RDD
dataHeader = NUMERIC_DATA_FILE.filter(lambda x: "Id" in x)

# Extract the data without headers. We can make use of the `subtract` function
dataNoHeader = NUMERIC_DATA_FILE.subtract(dataHeader)

numeric_temp_rdd = dataNoHeader.mapPartitions(lambda x: csv.reader(x, delimiter=","))
</code></pre>

<p>The Schema is passed in as a parameter into the <code>createDataFrame()</code> function</p>

<pre><code>numeric_df = sqlContext.createDataFrame(numeric_temp_rdd,schema)
numeric_df.registerTempTable("numeric")
</code></pre>

<p>Now, if you wish to convert this DataFrame to a Pandas dataframe, use the <code>toPandas()</code> function:</p>

<pre><code>pandas_df = numeric_df.limit(5).toPandas()
</code></pre>

<h2>The following statement will work as well:</h2>

<pre><code>numeric_df.select("Id")
</code></pre>

<h2>If you want to use pure SQL, you need to use SQLContext to query your table</h2>

<pre><code>from pyspark.sql import SQLContext
sqlContext  =  SQLContext(sc)
sqlContext.sql('SELECT Id from numeric')
</code></pre>

python pandas pyspark