title: PySpark XML to JSON w/ Time Series Data
body: <p>I have almost half a million XML files containing time series data that are about ~2-3MB each and contains around 10k rows of time series data per file. The idea is to convert the XML files into JSON for each unique ID. However, the time series data for each ID needs to be broken down into batches of row size 10 and converted to JSON and written to NoSQL database. Initially, the code was written to iterate over one monolithic dataframe for each ID and increment by row size 10 and then write document to db.</p>
  
  <pre><code>def resample_idx(X,resample_rate):
      for idx in range(0,len(X),resample_rate):
          yield X.iloc[idx:idx+resample_rate,:]
  
  # Batch Documents 
      for idx, df_batch in enumerate(resample_idx(df,10))
          dict_ = {}
          dict_['id'] = soup.find('id').contents[0]
          dict_['data'] = [v for k,v in pd.DataFrame.to_dict(df_batch.T).items()]
  </code></pre>
  
  <p>A sample of the JSON document looks as follows:</p>
  
  <pre><code>{'id':123456A,
  'data': [{'A': 251.23,
            'B': 130.56,
            'dtim': Timestamp('2011-03-24 11:18:13.350000')
           },
           {
            'A': 253.23,
            'B': 140.56,
            'dtim': Timestamp('2011-03-24 11:19:21.310000')
           },
           .........
          ]
  },
  {'id':123593X,
  'data': [{'A': 641.13,
            'B': 220.51,
            'C': 10.45
            'dtim': Timestamp('2011-03-26 12:11:13.350000')
           },
           {
            'A': 153.25,
            'B': 810.16,
            'C': 12.5
            'dtim': Timestamp('2011-03-26 12:11:13.310000')
           },
           .........
          ]
  }
  </code></pre>
  
  <p>This works fine for a small sample, but quickly realizing this will not scale when creating the batches. Therefore, looking to replicate this in Spark. Limited experience w/ Spark, but here's what I've attempted to far:</p>
  
  <p>First get all of time series data for all IDs:</p>
  
  <pre><code>df = sqlContext.read.format("com.databricks.spark.xml").options(rowTag='log').load("dbfs:/mnt/timedata/")
  </code></pre>
  
  <p>XML Schema</p>
  
  <pre><code> |-- _id: string (nullable = true)   
   |-- collect_list(TimeData): array (nullable = true)
   |    |-- element: struct (containsNull = true)
   |    |    |-- data: array (nullable = true)
   |    |    |    |-- element: string (containsNull = true)
   |    |    |-- ColNames: string (nullable = true)
   |    |    |-- Units: string (nullable = true)
  </code></pre>
  
  <p>SQL Query to get Spark DataFrame
      d = df.select("_id","TimeData.data",'TimeData.ColNames')</p>
  
  <p>Current Spark DataFrame</p>
  
  <pre><code>+--------------------+--------------------+--------------------+
  |                id  |                data|            ColNames|
  +--------------------+--------------------+--------------------+
  |123456A             |[2011-03-24 11:18...|dTim,A,B            |
  |123456A             |[2011-03-24 11:19...|dTim,A,B            |
  |123593X             |[2011-03-26 12:11...|dTim,A,B,C          |
  |123593X             |[2011-03-26 12:11...|dTim,A,B,C          |
  +--------------------+--------------------+--------------------+
  </code></pre>
  
  <p>Expected Spark DataFrame</p>
  
  <pre><code>+--------------------+--------------------+----------+----------+
  |                id  |               dTime|         A|         B|
  +--------------------+--------------------+----------+----------+
  |123456A             |2011-03-24 11:18... |    251.23|    130.56|
  |123456A             |2011-03-24 11:19... |    253.23|    140.56|
  +--------------------+--------------------+----------+----------+
  
  +--------------------+--------------------+----------+----------+----------+
  |                id  |               dTime|         A|         B|         C|
  +--------------------+--------------------+----------+----------+----------+
  |123593X             |2011-03-26 12:11... |    641.13|    220.51|     10.45|
  |123593X             |2011-03-26 12:11... |    153.25|    810.16|      12.5|
  +--------------------+-------------------+---------- +----------+----------+
  </code></pre>
  
  <p>I only showed data for two timestamps here, but how could I take the DataFrame above and turn into batch JSON files for every nth row (for each id), similarly to how it was done using Pandas shown above? Initially thoughts were performing a groupBy and apply a UDF to each ID? The output would look like the JSON structure above.</p>
  
  <p>XML Structure:</p>
  
  <pre><code>&lt;log&gt;
     &lt;id&gt;"ABC"&lt;/id&gt;
     &lt;TimeData&gt;
        &lt;colNames&gt;dTim,colA,colB,colC,&lt;/colNames&gt;
        &lt;data&gt;2011-03-24T11:18:13.350Z,0.139,38.988,0,110.307&lt;/data&gt;
        &lt;data&gt;2011-03-24T11:18:43.897Z,0.138,39.017,0,110.307&lt;/data&gt;
    &lt;/TimeData&gt;
  &lt;/log&gt;
  </code></pre>
  
  <p>Note that each ID does not have a fixed number of coNames and that can range between 5-30, depending on the data sources collected for that ID. </p>