title: How to loop through each row of pySpark DataFrames and write back to Hive
body: <p>I have some python code for hitting an API with records from a Hive Table and writing back to Hive as a new table with the additional columns from the API. I have a solution that works but unfortunately it is not scalable because of how long it takes.</p>
  <p>I think the issue is that I am using collect to loop over the hive table pulled in via Spark. I saw that to take advantage of Spark's parallel computation framework, you could define a custom function and use map but when I try that I get an error that I cannot use SQL Context within the map function. I am hoping someone can point me to the correct way to do this or at least a better way. Any feedback is appreciated, thank you!</p>
  <p>Steps in the code below:</p>
  <p>Pull in input table from Hive as Spark DataFrame</p>
  <p>For each row in the DataFrame:</p>
  <ol>
  <li>Generate an xml request with the row data</li>
  <li>Hit the API, receive and parse xml response</li>
  <li>Create a dictionary with input and output values</li>
  <li>Create Panda DataFrame from the dictionary</li>
  <li>Create Spark DataFrame from the Pandas DataFrame</li>
  <li>Write to CSV file in an HDFS directory that a Hive Table is built on</li>
  </ol>
  <pre><code>ResultsSpark = spark.sql(&quot;&quot;&quot;select * from my_input_table&quot;&quot;&quot;)
  
  for row in ResultsSpark.collect():
  
      input = row.input
  
      data = []
      cols = []
      
      data.append(input)
      cols.append(&quot;input&quot;)
          
      body = &quot;&quot;&quot;&lt;request&gt;&quot;&quot;&quot; + input + &quot;&quot;&quot;&lt;/request&gt;&quot;&quot;&quot;
      
      response = requests.post(url, data = body, headers = headers)
      
      root = ET.fromstring(response.content)
  
      
      
      ns = etree.FunctionNamespace('responseurl')
      ns.prefix = &quot;ns1&quot;
      ldoc = etree.parse(BytesIO(response.content))
      
      for sub_elements in ldoc.xpath('.//ns1:responseurl'):
          for element in sub_elements:
              data.append(element.text)
              cols.append(element.tag)
      
      df = pd.DataFrame(data).T  # Write in DF and transpose it
      df.columns = cols  # Update column names
      df.columns = df.columns.str.replace(&quot;responseurl&quot;, &quot;&quot;)
      
      schema = StructType([
          StructField(&quot;input&quot;,StringType(),True), 
          StructField(&quot;output&quot;, StringType(), True) \
      ])
      df_sdf = spark.createDataFrame(df, schema=schema)
  
      df_sdf.write.option(&quot;delimiter&quot;,&quot;^&quot;).save('hdfslocation', format='csv', mode='append')
  </code></pre>