I work on a dataframe with two column, mvv and count.

    +---+-----+
    |mvv|count|
    +---+-----+
    | 1 |  5  |
    | 2 |  9  |
    | 3 |  3  |
    | 4 |  1  |

i would like to obtain two list containing mvv values and count value. Something like

    mvv = [1,2,3,4]
    count = [5,9,3,1]
So, I tried the following code: The first line should return a python list of row. I wanted to see the first value:

    mvv_list = mvv_count_df.select(&#39;mvv&#39;).collect()
    firstvalue = mvv_list[0].getInt(0)

But I get an error message with the second line:

&gt; AttributeError: getInt



Convert spark DataFrame column to python list

I want to overwrite specific partitions instead of all in spark. I am trying the following command:

    df.write.orc(&#39;maprfs:///hdfs-base-path&#39;,&#39;overwrite&#39;,partitionBy=&#39;col4&#39;)

where df is dataframe having the incremental data to be overwritten.

hdfs-base-path contains the master data.

When I try the above command, it deletes all the partitions, and inserts those present in df at the hdfs path.

What my requirement is to overwrite only those partitions present in df at the specified hdfs path. Can someone please help me in this?

Overwrite specific partitions in spark dataframe write method

I created a dataframe in spark with the following schema:

    root
     |-- user_id: long (nullable = false)
     |-- event_id: long (nullable = false)
     |-- invited: integer (nullable = false)
     |-- day_diff: long (nullable = true)
     |-- interested: integer (nullable = false)
     |-- event_owner: long (nullable = false)
     |-- friend_id: long (nullable = false)

And the data is shown below:

    +----------+----------+-------+--------+----------+-----------+---------+
    |   user_id|  event_id|invited|day_diff|interested|event_owner|friend_id|
    +----------+----------+-------+--------+----------+-----------+---------+
    |   4236494| 110357109|      0|      -1|         0|  937597069|     null|
    |  78065188| 498404626|      0|       0|         0| 2904922087|     null|
    | 282487230|2520855981|      0|      28|         0| 3749735525|     null|
    | 335269852|1641491432|      0|       2|         0| 1490350911|     null|
    | 437050836|1238456614|      0|       2|         0|  991277599|     null|
    | 447244169|2095085551|      0|      -1|         0| 1579858878|     null|
    | 516353916|1076364848|      0|       3|         1| 3597645735|     null|
    | 528218683|1151525474|      0|       1|         0| 3433080956|     null|
    | 531967718|3632072502|      0|       1|         0| 3863085861|     null|
    | 627948360|2823119321|      0|       0|         0| 4092665803|     null|
    | 811791433|3513954032|      0|       2|         0|  415464198|     null|
    | 830686203|  99027353|      0|       0|         0| 3549822604|     null|
    |1008893291|1115453150|      0|       2|         0| 2245155244|     null|
    |1239364869|2824096896|      0|       2|         1| 2579294650|     null|
    |1287950172|1076364848|      0|       0|         0| 3597645735|     null|
    |1345896548|2658555390|      0|       1|         0| 2025118823|     null|
    |1354205322|2564682277|      0|       3|         0| 2563033185|     null|
    |1408344828|1255629030|      0|      -1|         1|  804901063|     null|
    |1452633375|1334001859|      0|       4|         0| 1488588320|     null|
    |1625052108|3297535757|      0|       3|         0| 1972598895|     null|
    +----------+----------+-------+--------+----------+-----------+---------+

I want to filter out the rows have null values in the field of &quot;friend_id&quot;.

    scala&gt; val aaa = test.filter(&quot;friend_id is null&quot;)

    scala&gt; aaa.count
I got :res52: Long = 0 which is obvious not right. What is the right way to get it?

One more question, I want to replace the values in the friend_id field. I want to replace null with 0 and 1 for any other value except null. The code I can figure out is:

    val aaa = train_friend_join.select($&quot;user_id&quot;, $&quot;event_id&quot;, $&quot;invited&quot;, $&quot;day_diff&quot;, $&quot;interested&quot;, $&quot;event_owner&quot;, ($&quot;friend_id&quot; != null)?1:0)

This code also doesn&#39;t work. Can anyone tell me how can I fix it? Thanks


how to filter out a null value from spark dataframe

I am new to Spark and Spark SQL.

How does `createOrReplaceTempView` work in Spark? 

If we register an `RDD` of objects as a table will spark keep all the data in memory? 

How does createOrReplaceTempView work in Spark?

I am using pyspark to read a parquet file like below:

    my_df = sqlContext.read.parquet(&#39;hdfs://myPath/myDB.db/myTable/**&#39;)

Then when I do `my_df.take(5)`, it will show `[Row(...)]`, instead of a table  format like when we use the pandas data frame.

Is it possible to display the data frame in a table format like pandas data frame? Thanks!

Pyspark: display a spark data frame in a table format

I have this code:

    l = [(&#39;Alice&#39;, 1),(&#39;Jim&#39;,2),(&#39;Sandra&#39;,3)]
    df = sqlContext.createDataFrame(l, [&#39;name&#39;, &#39;age&#39;])
    df.withColumn(&#39;age2&#39;, df.age + 2).toPandas()

Works fine, does what it needs to. Suppose though I only want to display the first n rows, and then call `toPandas()` to return a pandas dataframe. How do I do it? I can&#39;t call `take(n)` because that doesn&#39;t return a dataframe and thus I can&#39;t pass it to `toPandas()`.

So to put it another way, how can I take the top n rows from a dataframe and call `toPandas()` on the resulting dataframe? Can&#39;t think this is difficult but I can&#39;t figure it out.

I&#39;m using Spark 1.6.0.

Take n rows from a spark dataframe and pass to toPandas()

I&#39;m trying to convert Pandas DF into Spark one.
DF head:


    10000001,1,0,1,12:35,OK,10002,1,0,9,f,NA,24,24,0,3,9,0,0,1,1,0,0,4,543
    10000001,2,0,1,12:36,OK,10002,1,0,9,f,NA,24,24,0,3,9,2,1,1,3,1,3,2,611
    10000002,1,0,4,12:19,PA,10003,1,1,7,f,NA,74,74,0,2,15,2,0,2,3,1,2,2,691

Code:

    dataset = pd.read_csv(&quot;data/AS/test_v2.csv&quot;)
    sc = SparkContext(conf=conf)
    sqlCtx = SQLContext(sc)
    sdf = sqlCtx.createDataFrame(dataset)

And I got an error:

    TypeError: Can not merge type &lt;class &#39;pyspark.sql.types.StringType&#39;&gt; and &lt;class &#39;pyspark.sql.types.DoubleType&#39;&gt;



Converting Pandas dataframe into Spark dataframe error

I have the following sample DataFrame:


    a    | b    | c   | 
    
    1    | 2    | 4   |
    0    | null | null| 
    null | 3    | 4   |

And I want to replace null values only in the first 2 columns - Column &quot;a&quot; and &quot;b&quot;:

    a    | b    | c   | 
    
    1    | 2    | 4   |
    0    | 0    | null| 
    0    | 3    | 4   |

Here is the code to create sample dataframe: 

    rdd = sc.parallelize([(1,2,4), (0,None,None), (None,3,4)])
    df2 = sqlContext.createDataFrame(rdd, [&quot;a&quot;, &quot;b&quot;, &quot;c&quot;])

I know how to replace all null values using: 

    df2 = df2.fillna(0)

And when I try this, I lose the third column:
 
    df2 = df2.select(df2.columns[0:1]).fillna(0)





PySpark: How to fillna values in dataframe for specific columns?

Consider I have a defined schema for loading 10 csv files in a folder. Is there a way to automatically load tables using Spark SQL. I know this can be performed by using an individual dataframe for each file [given below], but can it be automated with a single command rather than pointing a file can I point a folder?

    df = sqlContext.read
           .format(&quot;com.databricks.spark.csv&quot;)
           .option(&quot;header&quot;, &quot;true&quot;)
           .load(&quot;../Downloads/2008.csv&quot;)



How to import multiple csv files in a single load?

I looked at the docs and it says the following join types are supported: 

&gt; Type of join to perform. Default inner. Must be one of: inner, cross,
&gt; outer, full, full_outer, left, left_outer, right, right_outer,
&gt; left_semi, left_anti.

I looked at the [StackOverflow answer](https://stackoverflow.com/questions/17946221/sql-join-and-different-types-of-joins) on SQL joins and top couple of answers do not mention some of the joins from above e.g. `left_semi` and `left_anti`. What do they mean in Spark?

What are the various join types in Spark?

I have a spark data frame `df`. Is there a way of sub selecting a few columns using a list of these columns?

    scala&gt; df.columns
    res0: Array[String] = Array(&quot;a&quot;, &quot;b&quot;, &quot;c&quot;, &quot;d&quot;)

I know I can do something like `df.select(&quot;b&quot;, &quot;c&quot;) `. But suppose I have  a list containing a few column names `val cols = List(&quot;b&quot;, &quot;c&quot;)`, is there a way to pass this to df.select? `df.select(cols)` throws an error. Something like `df.select(*cols)` as in python

Upacking a list to select multiple columns from a spark data frame

I am writing a User Defined Function which will take all the columns except the first one in a dataframe and do sum (or any other operation). Now the dataframe can sometimes have 3 columns or 4 columns or more. It will vary. 

I know I can hard code 4 column names as pass in the UDF but in this case it will vary so I would like to know how to get it done?

Here are two examples in the first one we have two columns to add and in the second one we have three columns to add.

[![enter image description here][1]][1]


  [1]: https://i.stack.imgur.com/ZVQ72.png

Pyspark: Pass multiple columns in UDF

I am trying to leverage spark partitioning. I was trying to do something like

    data.write.partitionBy(&quot;key&quot;).parquet(&quot;/location&quot;)

The issue here each partition creates huge number of parquet files which result slow read if I am trying to read from the root directory.

To avoid that I tried

    data.coalese(numPart).write.partitionBy(&quot;key&quot;).parquet(&quot;/location&quot;)

This however creates numPart number of parquet files in each partition.
Now my partition size is different. SO I would ideally like to have separate coalesce per partition. This is however doesn&#39;t look like an easy thing. I need to visit all the partition coalesce to a certain number and store at a separate location.

How should I use partitioning to avoid many files after write?


Spark parquet partitioning : Large number of files

How do I compute the cumulative sum per group specifically using the `DataFrame` `abstraction`; and in `PySpark`?

With an example dataset as follows:

    df = sqlContext.createDataFrame( [(1,2,&quot;a&quot;),(3,2,&quot;a&quot;),(1,3,&quot;b&quot;),(2,2,&quot;a&quot;),(2,3,&quot;b&quot;)], 
                                     [&quot;time&quot;, &quot;value&quot;, &quot;class&quot;] )

    +----+-----+-----+
    |time|value|class|
    +----+-----+-----+
    |   1|    2|    a|
    |   3|    2|    a|
    |   1|    3|    b|
    |   2|    2|    a|
    |   2|    3|    b|
    +----+-----+-----+

I would like to add a cumulative sum column of `value` for each `class` grouping over the (ordered) `time` variable.


Python Spark Cumulative Sum by Group Using DataFrame

Using Spark 1.6.1 version I need to fetch distinct values on a column and then perform some specific transformation on top of it. The column contains more than 50 million records and can grow larger.  
I understand that doing a `distinct.collect()` will bring the call back to the driver program. Currently I am performing this task as below, is there a better approach?

     import sqlContext.implicits._
     preProcessedData.persist(StorageLevel.MEMORY_AND_DISK_2)

     preProcessedData.select(ApplicationId).distinct.collect().foreach(x =&gt; {
       val applicationId = x.getAs[String](ApplicationId)
       val selectedApplicationData = preProcessedData.filter($&quot;$ApplicationId&quot; === applicationId)
       // DO SOME TASK PER applicationId
     })

     preProcessedData.unpersist()  


Fetching distinct values on a column using Spark DataFrame

I have a data frame in python/pyspark with columns `id` `time` `city` `zip` and so on......

Now I added a new column `name` to this data frame.

Now I have to arrange the columns in such a way that the `name` column comes after `id`

I have done like below

    change_cols = [&#39;id&#39;, &#39;name&#39;]

    cols = ([col for col in change_cols if col in df] 
            + [col for col in df if col not in change_cols])

    df = df[cols]

I am getting this error

    pyspark.sql.utils.AnalysisException: u&quot;Reference &#39;id&#39; is ambiguous, could be: id#609, id#1224.;&quot;


Why is this error occuring. How can I rectify this.

All Spark Dataframe Solutions on Gang of Coders

Total of 16 Spark Dataframe Solutions