title: Spark job not parallelising locally (using Parquet + Avro from local filesystem)
body: <p><strong>edit 2</strong></p>
  
  <p>Indirectly solved the problem by repartitioning the RDD into 8 partitions. Hit a roadblock with avro objects not being "java serialisable" found a snippet <a href="https://gist.github.com/massie/7224868" rel="nofollow">here to delegate avro serialisation to kryo.</a> The original problem still remains.</p>
  
  <p><strong>edit 1:</strong> Removed local variable reference in map function</p>
  
  <p>I'm writing a driver to run a compute heavy job on spark using parquet and avro for io/schema. I can't seem to get spark to use all my cores. What am I doing wrong ? Is it because I have set the keys to null ? </p>
  
  <p>I am just getting my head around how hadoop organises files. AFAIK since my file has a gigabyte of raw data I should expect to see things parallelising with the default block and page sizes. </p>
  
  <p>The function to ETL my input for processing looks as follows : </p>
  
  <pre><code>def genForum {
      class MyWriter extends AvroParquetWriter[Topic](new Path("posts.parq"), Topic.getClassSchema) {
        override def write(t: Topic) {
          synchronized {
            super.write(t)
          }
        }
      }
  
      def makeTopic(x: ForumTopic): Topic = {
        // Ommited to save space
      }
  
      val writer = new MyWriter
  
      val q =
        DBCrawler.db.withSession {
          Query(ForumTopics).filter(x =&gt; x.crawlState === TopicCrawlState.Done).list()
        }
  
      val sz = q.size
      val c = new AtomicInteger(0)
  
      q.par.foreach {
        x =&gt;
          writer.write(makeTopic(x))
          val count = c.incrementAndGet()
          print(f"\r${count.toFloat * 100 / sz}%4.2f%%")
      }
      writer.close()
    }
  </code></pre>
  
  <p>And my transformation looks as follows : </p>
  
  <pre><code>def sparkNLPTransformation() {
      val sc = new SparkContext("local[8]", "forumAddNlp")
  
      // io configuration
      val job = new Job()
      ParquetInputFormat.setReadSupportClass(job, classOf[AvroReadSupport[Topic]])
      ParquetOutputFormat.setWriteSupportClass(job,classOf[AvroWriteSupport])
      AvroParquetOutputFormat.setSchema(job, Topic.getClassSchema)
  
  
      // configure annotator
      val props = new Properties()
      props.put("annotators", "tokenize,ssplit,pos,lemma,parse")
      val an = DAnnotator(props)
  
  
      // annotator function
      def annotatePosts(ann : DAnnotator, top : Topic) : Topic = {
        val new_p = top.getPosts.map{ x=&gt;
          val at = new Annotation(x.getPostText.toString)
          ann.annotator.annotate(at)
          val t = at.get(classOf[SentencesAnnotation]).map(_.get(classOf[TreeAnnotation])).toList
  
          val r = SpecificData.get().deepCopy[Post](x.getSchema,x)
          if(t.nonEmpty) r.setTrees(t)
          r
        }
        val new_t = SpecificData.get().deepCopy[Topic](top.getSchema,top)
        new_t.setPosts(new_p)
        new_t
      }
  
      // transformation
      val ds = sc.newAPIHadoopFile("forum_dataset.parq", classOf[ParquetInputFormat[Topic]], classOf[Void], classOf[Topic], job.getConfiguration)
      val new_ds = ds.map(x=&gt; ( null, annotatePosts(x._2) ) )
  
      new_ds.saveAsNewAPIHadoopFile("annotated_posts.parq",
        classOf[Void],
        classOf[Topic],
        classOf[ParquetOutputFormat[Topic]],
        job.getConfiguration
      )
    }
  </code></pre>