how to do a nested for-each loop with PySpark
<p>Imagine a large dataset (>40GB parquet file) containing value observations of thousands of variables as triples <strong>(variable, timestamp, value)</strong>.</p>

<p>Now think of a query in which you are just interested in a subset of 500 variables. And you want to retrieve the observations (values --> time series) for those variables for specific points in time (observation window or timeframe). Such having a start and end time. </p>

<p>Without distributed computing (Spark), you could code it like this:</p>

<pre><code>for var_ in variables_of_interest:
    for incident in incidents:

        var_df = df_all.filter(
            (df.Variable == var_)
            &amp; (df.Time &gt; incident.startTime)
            &amp; (df.Time &lt; incident.endTime))
</code></pre>

<p><strong>My question is:</strong> how to do that with Spark/PySpark? I was thinking of either:</p>

<ol>
<li>joining the incidents somehow with the variables and filter the dataframe afterward.</li>
<li>broadcasting the incident dataframe and use it within a map-function when filtering the variable observations (df_all).</li>
<li>use RDD.cartasian or RDD.mapParitions somehow (remark: the parquet file was saved partioned by variable).</li>
</ol>

<p>The expected output should be:</p>

<pre><code>incident1 --&gt; dataframe 1
incident2 --&gt; dataframe 2
...
</code></pre>

<p>Where dataframe 1 contains all variables and their observed values within the timeframe of incident 1 and dataframe 2 those values within the timeframe of incident 2.</p>

<p>I hope you got the idea.</p>

<p><strong>UPDATE</strong></p>

<p>I tried to code a solution based on idea #1 and the code from the answer given by zero323. Work's quite well, but I wonder how to aggregate/group it to the incident in the final step? I tried adding a sequential number to each incident, but then I got errors in the last step. Would be cool if you can review and/or complete the code. Therefore I uploaded sample data and the scripts. The environment is Spark 1.4 (PySpark):</p>

<ul>
<li>Incidents: <a href="http://matthias-heise.de/stackoverflow/incidents.csv" rel="nofollow">incidents.csv</a></li>
<li>Variable value observation data (77MB): <a href="http://matthias-heise.de/stackoverflow/parameters_sample.csv" rel="nofollow">parameters_sample.csv</a> (put it to HDFS)</li>
<li>Jupyter Notebook: <a href="http://matthias-heise.de/stackoverflow/nested_for_loop_optimized.ipynb" rel="nofollow">nested_for_loop_optimized.ipynb</a></li>
<li>Python Script: <a href="http://matthias-heise.eu/stackoverflow/nested_for_loop_optimized.py" rel="nofollow">nested_for_loop_optimized.py</a></li>
<li>PDF export of Script: <a href="http://matthias-heise.eu/stackoverflow/nested_for_loop_optimized.pdf" rel="nofollow">nested_for_loop_optimized.pdf</a></li>
</ul>

<p>Generally speaking only the first approach looks sensible to me. Exact joining strategy on the number of records and distribution but you can either create a top level data frame:</p>

<pre><code>ref = sc.parallelize([(var_, incident) 
    for var_ in variables_of_interest:
    for incident in incidents
]).toDF(["var_", "incident"])
</code></pre>

<p>and simply <code>join</code></p>

<pre><code>same_var = col("Variable") == col("var_")
same_time = col("Time").between(
    col("incident.startTime"),
    col("incident.endTime")
)

ref.join(df.alias("df"), same_var &amp;  same_time)
</code></pre>

<p>or perform joins against particular partitions:</p>

<pre><code>incidents_ = sc.parallelize([
   (incident, ) for incident in incidents
]).toDF(["incident"])

for var_ in variables_of_interest:
    df = spark.read.parquet("/some/path/Variable={0}".format(var_))
    df.join(incidents_, same_time)
</code></pre>

<p>optionally <a href="http://stackoverflow.com/a/37487575/1560062">marking one side as small enough to be broadcasted</a>.</p>

python apache-spark pyspark