title: Running Parallel Threads in a PySpark Job
body: <p>I'm trying to run parallel threads in a spark job. This works without a hitch when I run the python script from the cli, but my understanding is that is not really capitalizing on the EMR cluster parallel processing benefits. It does not actually save the data when I run as a spark job. I'm not even sure it's creating the spark dataframe when I run it as a spark job. </p>
  
  <p>I also tried doing using map instead of doing the parallel threads, but couldn't get that to work either. </p>
  
  <p>If I can't get the parallelism to work as a spark job, it seems like I might as well just run it on a single ec2 instance with the parallel threads. </p>
  
  <p>So the basic logic is this - </p>
  
  <ol>
  <li>Create spark context at top of script</li>
  <li>Inside a class - Pull list of files to process from an SQS queue </li>
  <li><p>Loop over list of files with the following method</p>
  
  <pre><code># this is run for 10 blocks of 10 files each across the EMR cluster in parallel
  
  def parquet_driver(self):
  
      max_threads = 20
      futures=[]
      pool = ThreadPoolExecutor(max_threads)
      i = 0
      total_files_processed = 0
      while total_files_processed &lt;= len(self.master_file_list):
          while i &lt; max_threads:
              print('Processing %s' % self.master_file_list[i])
  
              futures.append(pool.submit(self.convert_to_parquet,
                self.master_file_list[i]))
              i += 1
  
          for x in as_completed(futures):
              pass
  
          # add in i number of files to the total
          total_files_processed += i
  </code></pre></li>
  </ol>
  
  <p>Notice this is passing a file to a method called "convert_to_parquet". </p>
  
  <pre><code>def convert_to_parquet(self, file):
  
  log_file_name = file.split(':')[2].replace('.dat', '.log')
  logger = Logger(log_file_name).get()
  
  try:
      bucket = s3.Bucket(file.split(':')[0])
      file_name = file.split(':')[2]
      file_obj = bucket.Object(file.split(':')[1] + '/' + file.split(':')[2])
      partition_key = file.split(':')[2].split('.')[2]
      target_table = file.split(':')[2].split('.')[1]
      receipt_handle = file.split(':')[3]
      file_contents = file_obj.get()["Body"].read()
      if 'al1' not in file.split(':')[2]:
          logger.debug('Record type = %s, deleting from queue and returning ..' % target_table)
      else:
          logger.debug('Working on %s..' % target_table)
          app_name = file
          #sc = SparkContext(appName=app_name)
          print('Reading the following file from s3: %s' % file_name)
          print('Found the following file contents on s3: %s' % file_contents)
          rdd = sc.parallelize(file_contents.split('\n')).map(lambda line: line.split(','))
  
          # rdd = sc.textFile(csv_file).map(lambda line: line.split(','))
          # pd.read_csv(csv_file)
          sqlContext = sql.SQLContext(sc)
  
          if hasattr(rdd, "toDF"):
              df = rdd.toDF()
          else:
              spark = SparkSession
              df = rdd.toDF()
  
          logger.debug("Partitioning data to: {0}".format(partition_key))
  
          # Go to redshift and get the data definition
          metadata = self.build_df_definition('al1')
  
          if 'cycle_date' in metadata['columns']:
              metadata['columns'].remove('cycle_date')
          if 'log_timestamp' in metadata['columns']:
              metadata['columns'].remove('log_timestamp')
  
          cols = metadata['columns']
          data_types = metadata['data_types']
  
          for idx in range(0,len(cols)):
              col_str = '_' + str(int(idx) + 1)
              df_field_value = regexp_replace(df[col_str], '"', '')
              df = df.withColumn(cols[idx],df_field_value.cast(data_types[idx]))
  
          df = df.withColumn("cycle_date",lit(partition_key))
          # this field will be pushed to the sqs queue
          df = df.withColumn("log_timestamp",lit(self.log_timestamp))
  
          full_cols = cols
          full_cols.append('cycle_date')
          full_cols.append('log_timestamp')
          print(full_cols)
          ref_df = df.select(full_cols)
          ref_df.show()
  
          partitionby=['year','month','day']
          output='/opt/data/' + '/' + target_table
          s3_loc = 's3://&lt;bucket&gt;/&lt;prefix&gt;/' + target_table
          codec='snappy'
  
          ref_df.write.partitionBy(['cycle_date']).format("parquet").save(s3_loc, mode="append")
  
          #sc.stop()
  except Exception as e:
      logger.debug(e)
      traceback.print_exc()
      open("/opt/logs/dump.log","w").write(traceback.print_exc())
      exit()
  else:
      # Delete received message from queue
      sqs.delete_message(
          QueueUrl=self.queue_url,
          ReceiptHandle=receipt_handle
      )
      logger.debug('Received and deleted file: %s' % file)
  </code></pre>