title: Avro: convert UNION schema to RECORD schema
body: <p>I have auto-generated Avro schema for simple class hierarchy:</p>
  
  <pre><code>trait T {def name: String}
  case class A(name: String, value: Int) extends T
  case class B(name: String, history: Array[String]) extends T
  </code></pre>
  
  <p>It looks like this:</p>
  
  <pre><code> [{
    "name": "org.example.schema.raw.A",
    "type": "record",
    "fields": [{
      "name": "name",
      "type": "string"
    }, {
      "name": "value",
      "type": "int"
    }]
  }, {
    "name": "org.example.schema.raw.B",
    "type": "record",
    "fields": [{
      "name": "name",
      "type": "string"
    }, {
      "name": "history",
      "type": {
        "type": "array",
        "items": "string"
      }
    }]
  }]
  </code></pre>
  
  <p>This schema works well for reading data from JSON into <code>GenericRecord</code> using plain Avro API. Next thing I try to achieve is storing all such <code>GenericRecord</code> objects into single parquet file using <code>AvroParquetWriter</code>:</p>
  
  <pre><code>val writer = new AvroParquetWriter[GenericRecord](file, schema)
  writer.write(record)
  writer.close()
  </code></pre>
  
  <p>This code fails at the first line with </p>
  
  <pre><code>java.lang.IllegalArgumentException: Avro schema must be a record.
  at parquet.avro.AvroSchemaConverter.convert(AvroSchemaConverter.java:96)
  at parquet.avro.AvroParquetWriter.writeSupport(AvroParquetWriter.java:137)
  at parquet.avro.AvroParquetWriter.&lt;init&gt;(AvroParquetWriter.java:54)
  at parquet.avro.AvroParquetWriter.&lt;init&gt;(AvroParquetWriter.java:86)
  </code></pre>
  
  <p>No wonder, <a href="https://github.com/Parquet/parquet-mr/blob/master/parquet-avro/src/main/java/parquet/avro/AvroSchemaConverter.java#L72" rel="nofollow noreferrer">AvroSchemaConverter</a> contains following lines:</p>
  
  <pre><code>if (!avroSchema.getType().equals(Schema.Type.RECORD)) {
        throw new IllegalArgumentException("Avro schema must be a record.");
  }
  </code></pre>
  
  <p>And my schema type is UNION. Any ideas/help in mapping (merging) this UNION schema into RECORD schema or any other suggestions are extremely appreciated. </p>
  
  <p><strong>SOLUTION</strong></p>
  
  <p>1) Read JSON from input using union scheme into <code>GenericRecord</code>
  2) Get or create <code>AvroParquetWriter</code> for type:</p>
  
  <pre><code>val writer = writers.getOrElseUpdate(record.getType, new AvroParquetWriter[GenericRecord](getPath(record.getType), record.getShema)
  </code></pre>
  
  <p>3) Write record into file:</p>
  
  <pre><code>writer.write(record)
  </code></pre>
  
  <p>4) Close all writers when all data are consumed from input:</p>
  
  <pre><code>writers.values.foreach(_.close())
  </code></pre>
  
  <p>5) Load data from directory into Spark SQL DataFrame:</p>
  
  <pre><code>sqlContext.option("mergeSchema", "true").parquet("/tmp/data/")
  </code></pre>
  
  <p>6) Data can be processed or stored as is - it's already merged by Spark:</p>
  
  <pre><code>df.write.format("parquet").save("merged.parquet")
  </code></pre>