**Let me help you to clarify them, both are common to use and important!**

    def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]

This is `prototype` of join, **please carefully look at it**. For example,

    val rdd1 = sc.makeRDD(Array((&quot;A&quot;,&quot;1&quot;),(&quot;B&quot;,&quot;2&quot;),(&quot;C&quot;,&quot;3&quot;)),2)
    val rdd2 = sc.makeRDD(Array((&quot;A&quot;,&quot;a&quot;),(&quot;C&quot;,&quot;c&quot;),(&quot;D&quot;,&quot;d&quot;)),2)
     
    scala&gt; rdd1.join(rdd2).collect
    res0: Array[(String, (String, String))] = Array((A,(1,a)), (C,(3,c)))

All keys that will appear in the final result **is common to** rdd1 and rdd2. This is similar to `relation database operation INNER JOIN`.

**But cogroup is different**,

    def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]

as one key at least appear in either of the two rdds, it will appear in the final result, let me clarify it:

    val rdd1 = sc.makeRDD(Array((&quot;A&quot;,&quot;1&quot;),(&quot;B&quot;,&quot;2&quot;),(&quot;C&quot;,&quot;3&quot;)),2)
    val rdd2 = sc.makeRDD(Array((&quot;A&quot;,&quot;a&quot;),(&quot;C&quot;,&quot;c&quot;),(&quot;D&quot;,&quot;d&quot;)),2)
    
    scala&gt; var rdd3 = rdd1.cogroup(rdd2).collect
    res0: Array[(String, (Iterable[String], Iterable[String]))] = Array(
    (B,(CompactBuffer(2),CompactBuffer())), 
    (D,(CompactBuffer(),CompactBuffer(d))), 
    (A,(CompactBuffer(1),CompactBuffer(a))), 
    (C,(CompactBuffer(3),CompactBuffer(c)))
    )

This is very `similar` to `relation database operation FULL OUTER JOIN`, but **instead of flattening the result per line per record, it will give you the `iterable interface` to you**, the following operation is **up to you as convenient**!

Good Luck!

Spark docs is: http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.rdd.PairRDDFunctions

I was reading up on async/await and when `Task.Yield` might be useful and came across [this post.][1]  I had a question regarding the below from that post:

&gt; When you use async/await, there is no guarantee that the method you
&gt; call when you do await `FooAsync()` will actually run asynchronously.
&gt; The internal implementation is free to return using a completely
&gt; synchronous path.

This is a little unclear to me probably because the definition of asynchronous in my head is not lining up.

In my mind, since I do mainly UI dev, async code is code that does not run on the UI thread, but on some other thread.  I guess in the text I quoted, a method is not truly async if it blocks on any thread (even if it&#39;s a thread pool thread for example).

**Question:**

If I have a long running task that is CPU bound (let&#39;s say it is doing a lot of hard math), then running that task asynchronously must be blocking some thread right?  Something has to actually do the math.  If I await it then some thread is getting blocked.

What is an example of a truly asynchronous method and how would they actually work?  Are those limited to I/O operations which take advantage of some hardware capabilities so no thread is ever blocked?

  [1]: https://stackoverflow.com/questions/22645024/when-would-i-use-task-yield

Confusion regarding threads and if asynchronous methods are truly asynchronous in C#

Sometimes the default [standard activations](https://keras.io/activations/) like ReLU, tanh, softmax, ... and the [advanced activations](https://keras.io/layers/advanced-activations/) like LeakyReLU aren&#39;t enough. And it might also not be in [keras-contrib](https://github.com/farizrahman4u/keras-contrib).

How do you create your own activation function?

How do you create a custom activation function with Keras?

What&#39;s the difference between join and cogroup in Apache Spark? What&#39;s the use case for each method?

What&#39;s the difference between join and cogroup in Apache Spark

<p>What's the difference between join and cogroup in Apache Spark? What's the use case for each method?</p>


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 upgraded IntelliJ to version CE 2017.3 from a CE 2017.2.5, and now I cannot build my sbt project. The build gets stuck on the first step:

    &quot;Build: sync
      MyProjName: syncing...
       dump project structure from sbt&quot;

I have deleted the .idea directory and the the `MyProjName/project/target` directory. I did an sbt build from the command line, which worked.
Then I started IntelliJ again and tried to import `MyProjName/build.sbt` as a project. I have also logged out/in. Did this several times. The import never progresses past this step.

I work behind a firewall, but I have checked my proxy settings, and they are correct and unchanged from the previous version (if this matters).

How can I troubleshoot / fix this?

The sbt version is 0.13.6. Scala plugin 2017.3.10.

From the log:

    IDE: IntelliJ IDEA (build #IC-173.3727.127, 27 Nov 2017 09:32)
    OS: Windows 7 (6.1, amd64) 
    JRE: 1.8.0_152-release-1024-b6 (JetBrains s.r.o) 
    JVM: 25.152-b6 (OpenJDK 64-Bit Server VM) 



IntelliJ Cannot Import sbt Project

This question came up after reading the [Loom proposal][1], which describes an approach of implementing coroutines in the Java programming language.

Particularly this proposal says that to implement this feature in the language, additional JVM support will be required.

As I understand it there are already several languages on the JVM that have coroutines as part of their feature set such as Kotlin and Scala.

So how is this feature implemented without additional support and can it be implemented efficiently without it?


  [1]: http://cr.openjdk.java.net/~rpressler/loom/Loom-Proposal.html

How are coroutines implemented in JVM langs without JVM support?

I am trying to leverage Akka&#39;s finite state machine framework for my use case. I am working on a system that processes a request that goes through various states.

The request here is the application name that needs to be deployed along with the application it depends on:

    Request for application A -&gt; A is in a QUEUED state
    Discover A&#39;s dependency B -&gt; B is in a QUEUED state
    B is being processed -&gt; B is in a PROCESSING STATE
    A is being processed -&gt; A is in a PROCESSING STATE
    B is processed -&gt; B is in a DONE state
    A is processed -&gt; A is in a DONE state

For this I am initializing a finite state machine at discovery time. So `A`&#39;s FSM is created when the request comes in, `B`&#39;s FSM is initialized when `B` is discovered from one of the actors.

Do I initialize and pass the FSM instance to all the actors, and at the same time `tell` the FSM about the operations being performed on the data so that the state machine goes in the correct state?

Here is the relevant portion of the state machine:

    when(QUEUED, matchEvent(requestAccepted.class, MyApp.class,
        (requestAccepted, service) -&gt; 
        goTo(PROCESSING).replying(PROCESSING)));
    
    when(PROCESSING, matchEvent(completed.class, MyApp.class,
        (completed, service) -&gt; goTo(DONE).replying(DONE)));

    // During transitions, save states in the database.
    onTransition(matchState(PROCESSING, DONE, () -&gt; {
      nextStateData().setServiceStatus(DONE);
      databaseWriter.tell(nextStateData(), getSelf());

And here is an example from one of the actors processing the request:

    ProcessingActor extends AbstractActor {
    
        @Override
          public void onReceive(Object message) throws Throwable {
            if (message instanceof processApplication) {
             // process the app
             // Initialize FSM for the Application
             FSM myFSM = Props.create(MYFSM.class);
             myFSM.tell( new completed(processApplication.app)
        }

Is this the right way to initialize the state machine and use it? Or should the initialization happen in the constructor of the `ProcessingActor`? But in that case there wouldn&#39;t be one state machine per application (data).


Akka finite state machine instances

I&#39;ve read about monoid homomorphism from [Monoid Morphisms, Products, and Coproducts](http://blog.higher-order.com/blog/2014/03/19/monoid-morphisms-products-coproducts/) and could not understand 100%.  

The author says (emphasis original):

&gt; The `length` function maps from `String` to `Int` *while preserving the
&gt; monoid structure*. Such a function, that maps from one monoid to
&gt; another in such a preserving way, is called a *monoid homomorphism*. In
&gt; general, for monoids `M` and `N`, a homomorphism `f: M =&gt; N`, and all values
&gt; `x:M`, `y:M`, the following equations hold:  

&gt;     f(x |+| y) == (f(x) |+| f(y))

&gt;     f(mzero[M]) == mzero[N]

Does he mean that, since the datatypes `String` and `Int` are monoids, and the function `length` maps `String =&gt; Int` preserving the monoid structure (`Int` is a monoid), it is called monoid homomorphism, right?

What is monoid homomorphism exactly?

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?

Calling `collect()` on an RDD will return the entire dataset to the driver which can cause out of memory and we should avoid that.

Will `collect()` behave the same way if called on a dataframe?&lt;br&gt;
What about the `select()` method?&lt;br&gt;
Does it also work the same way as `collect()` if called on a dataframe?

Spark dataframe: collect () vs select ()

    import numpy as np
    
    data = [
        (1, 1, None), 
        (1, 2, float(5)), 
        (1, 3, np.nan), 
        (1, 4, None), 
        (1, 5, float(10)), 
        (1, 6, float(&quot;nan&quot;)), 
        (1, 6, float(&quot;nan&quot;)),
    ]
    df = spark.createDataFrame(data, (&quot;session&quot;, &quot;timestamp1&quot;, &quot;id2&quot;))

**Expected output**

dataframe with count of nan/null for each column

***Note:***
The previous questions I found in stack overflow only checks for null &amp; not nan.
That&#39;s why I have created a new question.

I know I can use `isnull()` function in Spark to find number of Null values in Spark column but how to find Nan values in Spark dataframe?




How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?

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

We are reading data from MongoDB `Collection`. `Collection` column has two different values (e.g.: `(bson.Int64,int) (int,float)` ).

I am trying to get a datatype using pyspark.

My problem is some columns have different datatype.

Assume `quantity` and `weight` are the columns 

    quantity           weight
    ---------          --------
    12300              656
    123566000000       789.6767
    1238               56.22
    345                23
    345566677777789    21


Actually we didn&#39;t defined data type for any column of mongo collection.

When I query to the count from `pyspark dataframe`&lt;br/&gt;

    dataframe.count()

I got exception like this

	&quot;Cannot cast STRING into a DoubleType (value: BsonString{value=&amp;apos;200.0&amp;apos;})&quot;



Content Type	Original Author	Original Content on Stackoverflow
Question	miaoiao	View Question on Stackoverflow
Solution 1 - Scala	ashburshui	View Answer on Stackoverflow

What's the difference between join and cogroup in Apache Spark

Scala Problem Overview

Scala Solutions

Solution 1 - Scala

How do you create a custom activation function with Keras?

Confusion regarding threads and if asynchronous methods are truly asynchronous in C#

Attributions