title: Remove additional brackets from an RDD using PySpark
body: <p>I am doing aggregation on a set of fields using Spark in Python. I am taking few values as input &amp; creating an array out of it. Suppose i have 10 fields &amp; i am creating an array with 2 elements like first 9 fields as Key &amp; the 10 field as value &amp; passing this to the reduceByKey method to do the aggregation. </p>
  
  <p>The output i am getting has the array brackets, i am trying to eliminate those brackets before saving as a Textfile in HDFS. </p>
  
  <p>Here is the output i am getting</p>
  
  <pre><code>('11', '2015', '919810051877', '102', 'MOC', 'LOCAL', 'A2LL', 'H', '40336', '237', '1', '', '01', '102', '27/01/2016 01:05:36', 'hramcha'),48,0.8
  </code></pre>
  
  <p>I wanted it as,</p>
  
  <pre><code>'11', '2015', '919810051877', '102', 'MOC', 'LOCAL', 'A2LL', 'H', '40336', '237', '1', '', '01', '102', '27/01/2016 01:05:36', 'hramcha',48,0.8
  </code></pre>
  
  <p>Here's the code. I have tried lot of things but unable to eliminate the brackets.</p>
  
  <pre><code>    #! /usr/bin/env python
  
  import re
  import sys
  import time
  import decimal
  
  from pyspark import SparkContext
  from decimal import Decimal
  
  # function to extract the fields required for aggregation from the CDR files, based on position 
  def extractData(line):
      bal = line.strip()
      (MSISDN,calling_no,called_no,other_party_no,start_date,start_time,end_date,end_time,duration,call_direction,call_type,call_leg,usage_type,roam_type,
      IMSI,IMEI,technology,message_type,dialled_digits,first_LAC,first_cell_id,last_LAC,last_cell_id,Tt_file_no,Tt_date_time,SMSC_ID,GSM_CRN,IC_route,OG_route,
      carrier_code,location_routing_no,other_party_operator_id,MSC_id,MSC_circle,HOME_circle,pre_post_flag,disconnecting_party,disconnection_reason,
      internal_cause_LOC,MSRN,original_calling_no,original_called_no,A_party_country_code,ICR_operator,calling_no_port,called_no_port,translated_no,global_ref_no,
      GSM_SCF_control_of_AoC,SMS_result_code,level_of_camel_service,message_ref_no,network_ref_no,ori_method,type_of_called_subscriber,call_fwd_no,
      partial_indicator,CLI,call_identification_no,Irn_odbi_ind) = bal.split("|")
  
      time_s = []
      time_s = start_time.split(":")
      hr = time_s[0]
  
      load_date = time.strftime("%d/%m/%Y %H:%M:%S")
      load_user = "hramcha"
  
      return (start_date,hr,MSC_circle,HOME_circle,call_direction,call_type,call_leg,roam_type,first_cell_id,first_LAC,
      disconnection_reason,internal_cause_LOC,technology,pre_post_flag,load_date,load_user,duration)
  
  # function to remove brackets from RDD before saving as textfile
  def toCSVLine(data):
      return ','.join(str(d) for d in data)
  
  # function to create an array of (all the fields, call_duration)
  def formArray1(z):
      seg1 = z[0],z[1],z[2],z[3],z[4],z[5],z[6],z[7],z[8],z[9],z[10],z[11],z[12],z[13],z[14],z[15]
      seg2 = z[16]
      return seg1, seg2
  
  # function to create an array and also calculate mins_of_usage
  def calculateMins_of_usage(v):
      return v[0],v[1],float(v[1])/60
  
  
  # create Spark Context with the master details and the application name
  sc = SparkContext("spark://192.168.200.128:7077", "network_usage_HLY")
  
  # create an RDD from the input data in HDFS
  file = sc.textFile("hdfs://quickstart.cloudera/user/cloudera/input/network/",use_unicode=False) 
  
  # group by all the fields &amp; aggregate (sum) the call_duration
  data_aggregated = file.map(extractData).map(formArray1).reduceByKey(lambda a ,b: (int(a) + int(b)))
  
  # join the aggregation &amp; calculate mins_of_usage
  min_of_usage = data_aggregated.map(calculateMins_of_usage)
  
  # calling toCSVLine function to remove brackets &amp; braces
  lines = min_of_usage.map(toCSVLine)
  
  # save the final aggregated file in HDFS 
  lines.saveAsTextFile("hdfs://quickstart.cloudera/user/cloudera/output/network")
  </code></pre>