Pyspark: Multi-threaded Jobs
<p>I am having some problems with running multiple threads using a single SparkContext <code>sc</code>. I create a pool of threads and then each thread loads some data from the same table but using different rows. Each thread processes the loaded data under a dataframe and applies several transformations. </p>

<p>The current issue that I have is when I apply a transformation. For example I have a dataframe in the thread and then I apply a transformation like <code>StringIndexer(inputCol="category", "indexed").fit(df).transform(df)</code>. However, during the execution of those threads eventually I will receive an <code>IllegalArgumentException: u'requirement failed: Output column indexed already exists.</code> The weird thing about it is that if I turn the pool size to 1, then everything runs as it's supposed to. But, If I decide to increase the number of threads then it fails. It looks like the same dataframe is used on some of the threads. I have no idea why this is happening, I am not sharing any variables or anything between those threads (well only the  <code>SparkContext</code>).</p>

<p>For creating the threads I use the <code>multiprocessing.pool.ThreadPool</code>.</p>

<pre><code>def processModels(sc, table, models):
  for model in models:
    # Load the data coressponding to that model.
    query = "select * from %s where model=%d" % (table, model)
    df = sqlContext.sql(query)
    df = df.fillna(0).fillna("None").cache()

    # Apply some transformations to the dataset.        
    stages = []
    dtypes = df.dtypes 
    for dtype in dtypes:
        column_name = dtype[0]
        column_type = dtype[1]
        if(column_type=="string"):
            indexerModel = StringIndexer(inputCol=column_name, outputCol="%s_cat"%column_name).fit(df)
            if(len(indexerModel.labels)&gt;1):
                stages.append(indexerModel.copy())
    pipeline = PipelineModel(stages=stages)
    df = pipeline.transform(df)
    print("Model %d has %d samples." % (model, df.count()))
</code></pre>