I&#39;m just wondering what is the difference between an `RDD` and `DataFrame` *(Spark 2.0.0 DataFrame is a mere type alias for `Dataset[Row]`)* in Apache Spark? 

Can you convert one to the other?

Difference between DataFrame, Dataset, and RDD in Spark

I&#39;m trying to load an SVM file and convert it to a `DataFrame` so I can use the ML module (`Pipeline` ML) from Spark.
I&#39;ve just installed a fresh Spark 1.5.0 on an Ubuntu 14.04 (no `spark-env.sh` configured).


My `my_script.py` is:

    from pyspark.mllib.util import MLUtils
    from pyspark import SparkContext

    sc = SparkContext(&quot;local&quot;, &quot;Teste Original&quot;)
    data = MLUtils.loadLibSVMFile(sc, &quot;/home/svm_capture&quot;).toDF()

and I&#39;m running using:  `./spark-submit my_script.py`

And I get the error:

    Traceback (most recent call last):
    File &quot;/home/fred-spark/spark-1.5.0-bin-hadoop2.6/pipeline_teste_original.py&quot;, line 34, in &lt;module&gt;
    data = MLUtils.loadLibSVMFile(sc, &quot;/home/fred-spark/svm_capture&quot;).toDF()
    AttributeError: &#39;PipelinedRDD&#39; object has no attribute &#39;toDF&#39;

What I can&#39;t understand is that if I run:

    data = MLUtils.loadLibSVMFile(sc, &quot;/home/svm_capture&quot;).toDF()

directly inside PySpark shell, it works.




&#39;PipelinedRDD&#39; object has no attribute &#39;toDF&#39; in PySpark

I am looking for some better explanation of the aggregate functionality that is available via spark in python.  

The example I have is as follows (using pyspark from Spark 1.2.0 version)

    sc.parallelize([1,2,3,4]).aggregate(
      (0, 0),
      (lambda acc, value: (acc[0] + value, acc[1] + 1)),
      (lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])))

Output:

    (10, 4)


I get the expected result `(10,4)` which is sum of `1+2+3+4` and 4 elements.  If I change the initial value passed to the aggregate function to `(1,0)` from `(0,0)`  I get the following result

    sc.parallelize([1,2,3,4]).aggregate(
        (1, 0),
        (lambda acc, value: (acc[0] + value, acc[1] + 1)),
        (lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])))

Output:

    (19, 4)


The value increases by 9.  If I change it to `(2,0)`,  the value goes to `(28,4)` and so on.

Can someone explain to me how this value is calculated?  I expected the value to go up by 1 not by 9, expected to see `(11,4)` instead I am seeing `(19,4)`.  


Explain the aggregate functionality in Spark (with Python and Scala)

Assume `df1` and `df2` are two `DataFrame`s in Apache Spark, computed using two different mechanisms, e.g., Spark SQL vs. the Scala/Java/Python API. 

Is there an idiomatic way to determine whether the two data frames are equivalent (equal, isomorphic), where equivalence is determined by the data (column names and column values for each row) being identical save for the ordering of rows &amp; columns?

The motivation for the question is that there are often many ways to compute some big data result, each with its own trade-offs. As one explores these trade-offs, it is important to maintain correctness and hence the need to check for the equivalence/equality on a meaningful test data set.

DataFrame equality in Apache Spark

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

Trying to read a file located in S3 using spark-shell:

    scala&gt; val myRdd = sc.textFile(&quot;s3n://myBucket/myFile1.log&quot;)
    lyrics: org.apache.spark.rdd.RDD[String] = s3n://myBucket/myFile1.log MappedRDD[55] at textFile at &lt;console&gt;:12
    
    scala&gt; myRdd.count
    java.io.IOException: No FileSystem for scheme: s3n
    	at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:2607)
    	at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2614)
    	at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:91)
        ... etc ...

The **IOException: No FileSystem for scheme: s3n** error occurred with:

- Spark 1.31 or 1.40 on dev machine (no Hadoop libs)
- Running from the [Hortonworks Sandbox HDP v2.2.4](http://hortonworks.com/hdp/downloads/) (Hadoop 2.60) which integrates Spark 1.2.1 out of the box
- Using s3:// or s3n:// scheme

What is the cause of this error? Missing dependency, Missing configuration, or mis-use of `sc.textFile()`?

Or may be this is due to a bug that affects Spark build specific to Hadoop 2.60 as this [post](http://apache-spark-developers-list.1001551.n3.nabble.com/Spark-1-3-1-Hadoop-2-6-package-has-broken-S3-access-td12107.html) seems to suggest. I am going to try Spark for Hadoop 2.40 to see if this solves the issue.

Spark read file from S3 using sc.textFile (&quot;s3n://...)

How to give more column conditions when joining two dataframes. For example I want to run the following :

    val Lead_all = Leads.join(Utm_Master,  
        Leaddetails.columns(&quot;LeadSource&quot;,&quot;Utm_Source&quot;,&quot;Utm_Medium&quot;,&quot;Utm_Campaign&quot;) ==
        Utm_Master.columns(&quot;LeadSource&quot;,&quot;Utm_Source&quot;,&quot;Utm_Medium&quot;,&quot;Utm_Campaign&quot;),
    &quot;left&quot;)

I want to join only when these columns match. But above syntax is not valid as cols only takes one string. So how do I get what I want.


Spark specify multiple column conditions for dataframe join

In terms of `RDD` persistence, what are the differences between `cache()` and `persist()` in spark ? 



What is the difference between cache and persist?

I prefer Python over Scala. But, as Spark is natively written in Scala, I was expecting my code to run faster in the Scala than the Python version for obvious reasons.

With that assumption, I thought to learn &amp; write the Scala version of some very common preprocessing code for some 1&amp;nbsp;GB of data. Data is picked from the SpringLeaf competition on [Kaggle][1]. Just to give an overview of the data (it contains 1936 dimensions and 145232 rows). Data is composed of various types e.g. int, float, string, boolean. I am using 6 cores out of 8 for Spark processing; that&#39;s why I used `minPartitions=6` so that every core has something to process.

**Scala Code**

&lt;!-- language: scala --&gt;

    val input = sc.textFile(&quot;train.csv&quot;, minPartitions=6)

    val input2 = input.mapPartitionsWithIndex { (idx, iter) =&gt; 
      if (idx == 0) iter.drop(1) else iter }
    val delim1 = &quot;\001&quot;

    def separateCols(line: String): Array[String] = {
      val line2 = line.replaceAll(&quot;true&quot;, &quot;1&quot;)
      val line3 = line2.replaceAll(&quot;false&quot;, &quot;0&quot;)
      val vals: Array[String] = line3.split(&quot;,&quot;)

      for((x,i) &lt;- vals.view.zipWithIndex) {
        vals(i) = &quot;VAR_%04d&quot;.format(i) + delim1 + x
      }
      vals
    }

    val input3 = input2.flatMap(separateCols)

    def toKeyVal(line: String): (String, String) = {
      val vals = line.split(delim1)
      (vals(0), vals(1))
    }

    val input4 = input3.map(toKeyVal)

    def valsConcat(val1: String, val2: String): String = {
      val1 + &quot;,&quot; + val2
    }

    val input5 = input4.reduceByKey(valsConcat)

    input5.saveAsTextFile(&quot;output&quot;)

**Python Code**


&lt;!-- language: python --&gt;


    input = sc.textFile(&#39;train.csv&#39;, minPartitions=6)
    DELIM_1 = &#39;\001&#39;


    def drop_first_line(index, itr):
      if index == 0:
        return iter(list(itr)[1:])
      else:
        return itr

    input2 = input.mapPartitionsWithIndex(drop_first_line)

    def separate_cols(line):
      line = line.replace(&#39;true&#39;, &#39;1&#39;).replace(&#39;false&#39;, &#39;0&#39;)
      vals = line.split(&#39;,&#39;)
      vals2 = [&#39;VAR_%04d%s%s&#39; %(e, DELIM_1, val.strip(&#39;\&quot;&#39;))
               for e, val in enumerate(vals)]
      return vals2


    input3 = input2.flatMap(separate_cols)

    def to_key_val(kv):
      key, val = kv.split(DELIM_1)
      return (key, val)
    input4 = input3.map(to_key_val)

    def vals_concat(v1, v2):
      return v1 + &#39;,&#39; + v2

    input5 = input4.reduceByKey(vals_concat)
    input5.saveAsTextFile(&#39;output&#39;)

**Scala Performance**
Stage 0 (38 mins), Stage 1 (18 sec)
[![enter image description here][2]][2]

**Python Performance**
Stage 0 (11 mins), Stage 1 (7 sec)
[![enter image description here][3]][3]

Both produces different DAG visualization graphs (due to which both pictures show different stage 0 functions for Scala (`map`) and Python (`reduceByKey`))

But, essentially both code tries to transform data into (dimension_id, string of list of values) RDD and save to disk. The output will be used to compute various statistics for each dimension.

Performance wise, Scala code for this real data like this seems to run **4 times slower** than the Python version.
Good news for me is that it gave me good motivation to stay with Python. Bad news is I didn&#39;t quite understand why?


  [1]: https://en.wikipedia.org/wiki/Kaggle
  [2]: http://i.stack.imgur.com/uDJ30.png
  [3]: http://i.stack.imgur.com/3wwcQ.png

Spark performance for Scala vs Python

When a resilient distributed dataset (RDD) is created from a text file or collection (or from another RDD), do we need to call &quot;cache&quot; or &quot;persist&quot; explicitly to store the RDD data into memory? Or is the RDD data stored in a distributed way in the memory by default?

    val textFile = sc.textFile(&quot;/user/emp.txt&quot;)

As per my understanding, after the above step, textFile is a RDD and is available in all/some of the node&#39;s memory.

If so, why do we need to call &quot;cache&quot; or &quot;persist&quot; on textFile RDD then?

(Why) do we need to call cache or persist on a RDD

What&#39;s the difference between an [RDD&#39;s][1] `map` and `mapPartitions` method? And does `flatMap` behave like `map` or like `mapPartitions`? Thanks.

(edit)
i.e. what is the difference (either semantically or in terms of execution) between

      def map[A, B](rdd: RDD[A], fn: (A =&gt; B))
                   (implicit a: Manifest[A], b: Manifest[B]): RDD[B] = {
        rdd.mapPartitions({ iter: Iterator[A] =&gt; for (i &lt;- iter) yield fn(i) },
          preservesPartitioning = true)
      }

And:

      def map[A, B](rdd: RDD[A], fn: (A =&gt; B))
                   (implicit a: Manifest[A], b: Manifest[B]): RDD[B] = {
        rdd.map(fn)
      }

  [1]: https://spark.apache.org/docs/latest/api/scala/#org.apache.spark.rdd.RDD

Apache Spark: map vs mapPartitions?

How can I convert an RDD (`org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]`) to a Dataframe `org.apache.spark.sql.DataFrame`. I converted a dataframe to rdd using `.rdd`. After processing it I want it back in dataframe. How can I do this ?


How to convert rdd object to dataframe in spark

From my Spark UI. What does it mean by skipped?

[![enter image description here][1]][1]


  [1]: http://i.stack.imgur.com/cyvd1.png

What does &quot;Stage Skipped&quot; mean in Apache Spark web UI?

I read up on the documentation of [`HashPartitioner`][1]. Unfortunately nothing much was explained except for the API calls. I am under the assumption that `HashPartitioner` partitions the distributed set based on the hash of the keys. For example if my data is like 

    (1,1), (1,2), (1,3), (2,1), (2,2), (2,3)

So partitioner would put this into different partitions with same keys falling in the same partition. However I do not understand the significance of the constructor argument 

    new HashPartitoner(numPartitions) //What does numPartitions do?

For the above dataset how would the results differ if I did 

    new HashPartitoner(1)
    new HashPartitoner(2)
    new HashPartitoner(10)

So how does `HashPartitioner` work actually?

  [1]: http://spark.apache.org/docs/1.3.1/api/java/index.html?org/apache/spark/HashPartitioner.html

How does HashPartitioner work?

How can I find median of an `RDD` of integers using a distributed method, IPython, and Spark? The `RDD` is approximately 700,000 elements and therefore too large to collect and find the median.

This question is similar to this question. However, the answer to the question is using Scala, which I do not know.

https://stackoverflow.com/questions/28158729/how-can-i-calculate-exact-median-with-apache-spark

Using the thinking for the Scala answer, I am trying to write a similar answer in Python.

I know I first want to sort the `RDD`. I do not know how. I see the `sortBy` (Sorts this RDD by the given `keyfunc`) and `sortByKey` (Sorts this `RDD`, which is assumed to consist of (key, value) pairs.) methods. I think both use key value and my `RDD` only has integer elements.

1. First, I was thinking of doing `myrdd.sortBy(lambda x: x)`? 
2. Next I will find the length of the rdd (`rdd.count()`).
3. Finally, I want to find the element or 2 elements at the center of the rdd. I need help with this method too.

EDIT:

I had an idea. Maybe I can index my `RDD` and then key = index and value = element. And then I can try to sort by value? I don&#39;t know if this is possible because there is only a `sortByKey` method.

How to find median and quantiles using Spark

According to Learning Spark

&gt;Keep in mind that repartitioning your data is a fairly expensive operation.
Spark also has an optimized version of `repartition()` called `coalesce()` that allows avoiding data movement, but only if you are decreasing the number of RDD partitions.

One difference I get is that with `repartition()` the number of partitions can be increased/decreased, but with `coalesce()` the number of partitions can only be decreased.

If the partitions are spread across multiple machines and `coalesce()` is run, how can it avoid data movement?


Spark - repartition() vs coalesce()

I am new to Spark and Scala. I was confused about the way reduceByKey function works in Spark. Suppose we have the following code:

    val lines = sc.textFile(&quot;data.txt&quot;)
    val pairs = lines.map(s =&gt; (s, 1))
    val counts = pairs.reduceByKey((a, b) =&gt; a + b)

The map function is clear: s is the key and it points to the line from `data.txt` and 1 is the value.

However, I didn&#39;t get how the reduceByKey works internally? Does &quot;a&quot; points to the key? Alternatively, does &quot;a&quot; point to &quot;s&quot;? Then what does represent a + b? how are they filled?

reduceByKey: How does it work internally?

The [Spark research paper][1] has prescribed a new distributed programming model over classic Hadoop MapReduce, claiming the simplification and vast performance boost in many cases specially on Machine Learning. However, the material to uncover the `internal mechanics` on `Resilient Distributed Datasets` with `Directed Acyclic Graph` seems lacking in this paper.

Should it be better learned by investigating the source code?


  [1]: http://www.cs.berkeley.edu/~matei/papers/2012/nsdi_spark.pdf

How DAG works under the covers in RDD?

In Spark version *1.2.0* one could use `subtract` with 2 `SchemRDD`s to end up with only the different content from the first one

    val onlyNewData = todaySchemaRDD.subtract(yesterdaySchemaRDD)

`onlyNewData` contains the rows in `todaySchemRDD` that do not exist in `yesterdaySchemaRDD`.

How can this be achieved with `DataFrames` in Spark version *1.3.0*?

Spark: subtract two DataFrames

RDD has a _meaningful_ (as opposed to some random order imposed by the storage model) order if it was processed by [`sortBy()`](http://spark.apache.org/docs/latest/api/java/org/apache/spark/rdd/RDD.html#sortBy%28scala.Function1,%20boolean,%20int,%20scala.math.Ordering,%20scala.reflect.ClassTag%29), as explained in this [reply](https://stackoverflow.com/a/29281548/850781).

Now, which operations **preserve** that order?

E.g., is it ***guaranteed*** that (after `a.sortBy()`)

    a.map(f).zip(a) === 
    a.map(x =&gt; (f(x),x))

How about

    a.filter(f).map(g) === 
    a.map(x =&gt; (x,g(x))).filter(f(_._1)).map(_._2)

what about

    a.filter(f).flatMap(g) === 
    a.flatMap(x =&gt; g(x).map((x,_))).filter(f(_._1)).map(_._2)

Here &quot;equality&quot; `===` is understood as &quot;functional equivalence&quot;, i.e., there is no way to distinguish the outcome using user-level operations (i.e., without reading logs &amp;c).

All Rdd Solutions on Gang of Coders

Total of 20 Rdd Solutions