title: Combine two rows in spark based on a condition in pyspark
body: <p>I have input record in following format:
  <a href="https://i.stack.imgur.com/rwPmH.jpg" rel="nofollow noreferrer">Input data format</a></p>
  
  <p><a href="https://i.stack.imgur.com/jeW6N.jpg" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/jeW6N.jpg" alt="enter image description here"></a></p>
  
  <p>I want the data to be transofmmed in the following format:
  <a href="https://i.stack.imgur.com/qBF8r.jpg" rel="nofollow noreferrer">Output data format</a></p>
  
  <p><a href="https://i.stack.imgur.com/gFddc.jpg" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/gFddc.jpg" alt="enter image description here"></a></p>
  
  <p>I want to combine my 2 rows based on the condition type.</p>
  
  <p>As per my knowledge I need to take the composite key of the 3 data fields and compare the type fields once they are equal.</p>
  
  <p>Can someone please help me with the implementation in Spark using Python?</p>
  
  <p>EDIT:
  Following is my try using RDD in pyspark</p>
  
  <pre><code>record = spark.read.csv("wasb:///records.csv",header=True).rdd
  print("Total records: %d")%record.count()
  private_ip = record.map(lambda fields: fields[2]).distinct().count()
  private_port = record.map(lambda fields: fields[3]).distinct().count()
  destination_ip = record.map(lambda fields: fields[6]).distinct().count()
  destination_port = record.map(lambda fields: fields[7]).distinct().count()
  print("private_ip:%d, private_port:%d, destination ip:%d, destination_port:%d")%(private_ip,private_port,destination_ip,destination_port)
  types = record.map(lambda fields: ((fields[2],fields[3],fields[6],fields[7]),fields[0])).reduceByKey(lambda a,b:a+','+b)
  print types.first()
  </code></pre>
  
  <p>And following is my output till now. </p>
  
  <pre><code>((u'100.79.195.101', u'54835', u'58.96.162.33', u'80'), u'22-02-2016 13:11:03,22-02-2016 13:13:53')
  </code></pre>