Myujth

I'm reading in hundreds of XML files into a Spark Dataframe, where each row consist of meta data and time series data for a particular event. Each one of these rows is converted into a rdd to be transformed into batches of documents with a particular key/vale structure and then written over to a database. The XML data need to be broken into batches <50Kb, hence the helper function to produce then batches shown below.

def build_documents(data):

 # Make dataframe out of data tags
 data = pd.DataFrame([i.split(',') for i in list(chain(*(data)))])

 # Helper function to Get Batches
 for batch in get_batches(data): 
 x = batch.T.to_dict()
 yield x

def process_partition(partition):
 client = document_client.DocumentClient(HOST, 'masterKey': MASTER_KEY )
 for element in partition:
 generator = build_documents(element)
 for batch in generator:
 client.CreateDocument(collection_link + 'data', batch)


# Write to Database
df.rdd.coalesce(20).foreachPartition(process_partition)

Still tuning the number of partitions, but any thoughts on how this can be improved? Performance is really slow, as expected with the code implemented so far. The cluster consist of 32 cores, 128.0 GB Memory for both driver and can scale up to 8 executors. As shown below, there is only two workers running, which obviously is not optimal when scaling up further. Thoughts?

df.rdd.coalesce(20).foreachPartition(process_partition) will write sequential entries to database. and morever your logic for function process_partition will also be sequential.

You need to multithread the logic for def process_partition. That will speed up the process. Also use df.rdd.coalesce(20).foreachPartitionAsync(process_partition)