title: Higher-order functions in pyspark to generate streaming dataframe operations?
body: <p>Let's say I've got a list of a lot of Kafka topics to listen to and table names to upsert into. So, I write this:</p>
  
  <pre><code>def upsert(df, tableName):
    df.createOrReplaceTempView("mergeTable")
    spark.sql(f'''
  MERGE INTO {tableName}
  USING mergeTable
  ON {tableName}.Key = mergeTable.Key
  WHEN MATCHED THEN
  UPDATE SET
  {tableName}.Value = mergeTable.Value
  WHEN NOT MATCHED
  THEN INSERT (Key, Value) VALUES
  (mergeTable.Key, mergeTable.Value)
    ''')
  def curryUpsert(df, tableName):
    def F(df, batchId):
      return upsert(df,tableName)
    return F
  </code></pre>
  
  <p>and then I call it like this:</p>
  
  <pre><code>for (topic, tableName) in a_big_list_of_tables_to_upsert_into:
    (
        spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", bootstrapServers)
        .option("subscribe", topic)
        .load()  
        .writeStream
        .foreachBatch(curryUpsert(tableName))
        .outputMode("update")
        .start()
      )
  </code></pre>
  
  <p>I get this error: <code>Exception: It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transformation.</code></p>
  
  <p>Is there a way to create a spark streaming dataframe generator? I want to fan a collection out into an equal number of streaming datareaders. I could do this by writing each one by hand, but it seems like I should be able to write a function that spits them out for me, but spark magic keeps that from working for me.</p>