title: Saving Spark RDD to Avro with spark.api.python.Converter
body: <p>I am trying to follow <a href="https://databricks.com/blog/2014/09/17/spark-1-1-bringing-hadoop-inputoutput-formats-to-pyspark.html" rel="nofollow">Spark 1.1: Bringing Hadoop Input/Output Formats to PySpark</a> guide to persist RDD in Avro format. My Converter works but looks really ugly.</p>
  
  <pre><code>class JavaToAvroWrapperConverter()
    extends Converter[Any, AvroKey[GenericData.Record]] {
    override def convert(obj: Any): AvroKey[GenericData.Record] = {
      val args: Array[Any] = obj.asInstanceOf[Array[Any]]
      val schema = (new Parser).parse(args(1).asInstanceOf[String])
      val map = args(0).asInstanceOf[Array[Any]](0).asInstanceOf[util.HashMap[String, Any]]
      val record = new GenericData.Record(schema)
      map.keys.foreach(k =&gt; record.put(k, map(k)))
  
      new AvroKey[GenericData.Record](record)
    }
  }
  </code></pre>
  
  <p>for that to work I need to construct my RDD as a map of each row to the schema</p>
  
  <pre><code>to_avro_rdd.map(lambda x: ((x, avroSchema), None)).saveAsNewAPIHadoopFile(
          path + ".out",
          keyClass = "org.apache.avro.mapred.AvroKey",
          valueClass = "org.apache.hadoop.io.NullWritable",
          outputFormatClass = "org.apache.avro.mapreduce.AvroKeyOutputFormat",
          keyConverter="org.apache.spark.examples.pythonconverters.JavaToAvroWrapperConverter",
          conf=conf
      )
  </code></pre>
  
  <p>So, my question is whether there is a more civilized way to handle that without repeating schema in every RDD tuple and than getting it in a form of bulky parameters in converter. Or a short version of the same question would be: what am I doing wrong?</p>
  
  <p>Thanks!</p>