is there any way to split Spark Dataset in given logic
<p>i am looking for Spark Dataset split application which is similar to bellow mentioned logic. </p>

<pre><code>&gt;&gt;&gt; import pandas as pd
&gt;&gt;&gt; import numpy as np

&gt;&gt;&gt; df1 = pd.DataFrame(np.random.randn(10, 4), columns=['a', 'b', 'c', 'd'])
&gt;&gt;&gt; df1
          a         b         c         d
0 -0.398502 -1.083682  0.850632 -1.443868
1 -2.124333  1.093590 -0.338162 -1.414035
2  0.753560  0.600687 -0.998277 -2.094359
3 -0.635962 -0.291226  0.428961  1.158153
4 -0.900506 -0.545226 -0.448178 -0.567717
5  0.112911  0.351649  0.788940  2.071541
6 -0.358625  0.500367  1.009819 -1.139761
7  1.003608  0.246925  0.225138 -0.586727
8  0.355274 -0.540685  1.482472  0.364989
9  3.089610 -1.415088 -0.072107 -0.203137
&gt;&gt;&gt;

&gt;&gt;&gt; mask = df1.applymap(lambda x: x &lt;-0.7)
&gt;&gt;&gt;
&gt;&gt;&gt; mask
       a      b      c      d
0  False   True  False   True
1   True  False  False   True
2  False  False   True   True
3  False  False  False  False
4   True  False  False  False
5  False  False  False  False
6  False  False  False   True
7  False  False  False  False
8  False  False  False  False
9  False   True  False  False
&gt;&gt;&gt; mask.any(axis=1)
0     True
1     True
2     True
3    False
4     True
5    False
6     True
7    False
8    False
9     True
dtype: bool
&gt;&gt;&gt; df1 = df1[-mask.any(axis=1)]
&gt;&gt;&gt; df1
          a         b         c         d
3 -0.635962 -0.291226  0.428961  1.158153
5  0.112911  0.351649  0.788940  2.071541
7  1.003608  0.246925  0.225138 -0.586727
8  0.355274 -0.540685  1.482472  0.364989
&gt;&gt;&gt;
</code></pre>

<p>In spark i gone though <code>df.filter</code> but its trying pick only matched ,  but in my case i need to filter(remove) data into 3 -4 level. only one level i shown above.  which is kind of filtering. </p>

<p>Preserving order is very difficult in Spark applications due to the assumptions of the RDD abstraction. The best approach you can take is to translate the pandas logic using the Spark api, like I've done here. Unfortunately, I do not think you can apply the same filter criteria to every column, so I had to manually translate the mask into operations on multiple columns. This <a href="https://databricks.com/blog/2015/08/12/from-pandas-to-apache-sparks-dataframe.html" rel="nofollow">Databricks blog post</a> is helpful for anyone transitioning from Pandas to Spark. </p>

<pre><code>import pandas as pd
import numpy as np
np.random.seed(1000)
df1 = pd.DataFrame(np.random.randn(10, 4), columns=['a', 'b', 'c', 'd'])
mask = df1.applymap(lambda x: x &lt;-0.7)
df2 = df1[-mask.any(axis=1)]
</code></pre>

<p>The result we want is: </p>

<pre><code>          a         b         c         d
1 -0.300797  0.389475 -0.107437 -0.479983
5 -0.334835 -0.099482  0.407192  0.919388
6  0.312118  1.533161 -0.550174 -0.383147
8 -0.326925 -0.045797 -0.304460  1.923010
</code></pre>

<p>So in Spark, we create the dataframe using the Pandas data frame and use <code>filter</code> to get the correct result set: </p>

<pre><code>df1_spark = sqlContext.createDataFrame(df1).repartition(10)
df2_spark = df1_spark.filter(\
   (df1_spark.a &gt; -0.7)\
 &amp; (df1_spark.b &gt; -0.7)\
 &amp; (df1_spark.c &gt; -0.7)\
 &amp; (df1_spark.d &gt; -0.7)\
 )
</code></pre>

<p>Which gives us the proper result (notice the order is not preserved): </p>

<pre><code>df2_spark.show()
+-------------------+--------------------+--------------------+-------------------+
|                  a|                   b|                   c|                  d|
+-------------------+--------------------+--------------------+-------------------+
|-0.3348354532115408| -0.0994816980097769| 0.40719210034152314|  0.919387539204449|
| 0.3121180100663634|  1.5331610653579348| -0.5501738650283003|-0.3831474108842978|
|-0.3007966727870205|  0.3894745542873072|-0.10743730169089667|-0.4799830753607686|
| -0.326924675176391|-0.04579718800728687| -0.3044600616968845|  1.923010130400007|
+-------------------+--------------------+--------------------+-------------------+
</code></pre>

<p>If you <strong><em>absolutely needed</em></strong> to create the mask using Pandas, you would have to preserve the index of the original Pandas dataframe and remove individual records from the Spark by creating a broadcast variable and filtering based on the index column. Here's an example, YMMV. </p>

<p>Add an index: </p>

<pre><code>df1['index_col'] = df1.index

df1
          a         b         c         d  index_col
0 -0.804458  0.320932 -0.025483  0.644324          0
1 -0.300797  0.389475 -0.107437 -0.479983          1
2  0.595036 -0.464668  0.667281 -0.806116          2
3 -1.196070 -0.405960 -0.182377  0.103193          3
4 -0.138422  0.705692  1.271795 -0.986747          4
5 -0.334835 -0.099482  0.407192  0.919388          5
6  0.312118  1.533161 -0.550174 -0.383147          6
7 -0.822941  1.600083 -0.069281  0.083209          7
8 -0.326925 -0.045797 -0.304460  1.923010          8
9 -0.078659 -0.582066 -1.617982  0.867261          9
</code></pre>

<p>Convert the mask into a Spark broadcast variable: </p>

<pre><code>myIdx = sc.broadcast(df2.index.tolist())
</code></pre>

<p>Create and modify the dataframes using the Spark api: </p>

<pre><code>df1_spark.rdd.filter(lambda row: row and row['index_col'] not in myIdx.value).collect()
df2_spark = df1_spark.rdd.filter(lambda row: row and row['index_col'] in myIdx.value).toDF()

df2_spark.show()
+-------------------+--------------------+--------------------+-------------------+---------+
|                  a|                   b|                   c|                  d|index_col|
+-------------------+--------------------+--------------------+-------------------+---------+
|-0.3007966727870205|  0.3894745542873072|-0.10743730169089667|-0.4799830753607686|        1|
|-0.3348354532115408| -0.0994816980097769| 0.40719210034152314|  0.919387539204449|        5|
| 0.3121180100663634|  1.5331610653579348| -0.5501738650283003|-0.3831474108842978|        6|
| -0.326924675176391|-0.04579718800728687| -0.3044600616968845|  1.923010130400007|        8|
+-------------------+--------------------+--------------------+-------------------+---------+
</code></pre>

python pandas apache-spark