diff --git a/docs/development/extensions-core/avro.md b/docs/development/extensions-core/avro.md
index a200e83..75704cf 100644
--- a/docs/development/extensions-core/avro.md
+++ b/docs/development/extensions-core/avro.md
@@ -39,13 +39,30 @@
 
 Druid supports most Avro types natively, there are however some exceptions which are detailed here.
 
-`union` types which aren't of the form `[null, otherType]` aren't supported at this time.
+#### Unions
+Druid has two modes for supporting `union` types.
 
+The default mode will treat unions as a single value regardless of the type it is populated with.
+
+If you wish to operate on each different member of a union however you can set `extractUnionsByType` on the Avro parser in which case unions will be expanded into nested objects according to the following rules:
+* Primitive types and unnamed complex types are keyed their type name. i.e `int`, `string`
+* Complex named types are keyed by their names, this includes `record`, `fixed` and `enum`.
+* The Avro null type is elided as its value can only ever be null
+
+This is safe because an Avro union can only contain a single member of each unnamed type and duplicates of the same named type are not allowed.
+i.e only a single array is allowed, multiple records (or other named types) are allowed as long as each has a unique name.
+
+The members can then be accessed using a [flattenSpec](../../ingestion/data-formats.md#flattenspec) similar other nested types.
+
+#### Binary types
 `bytes` and `fixed` Avro types will be returned by default as base64 encoded strings unless the `binaryAsString` option is enabled on the Avro parser.
 This setting will decode these types as UTF-8 strings.
 
+#### Enums
 `enum` types will be returned as `string` of the enum symbol.
 
+#### Complex types
 `record` and `map` types representing nested data can be ingested using [flattenSpec](../../ingestion/data-formats.md#flattenspec) on the parser.
 
-Druid doesn't currently support Avro logical types, they will be ignored and fields will be handled according to the underlying primitive type.
\ No newline at end of file
+#### Logical types
+Druid doesn't currently support Avro logical types, they will be ignored and fields will be handled according to the underlying primitive type.
diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java
index 39ce48e..165f6a4 100644
--- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java
+++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java
@@ -34,18 +34,24 @@
 {
   private final ParseSpec parseSpec;
   private final boolean fromPigAvroStorage;
+  private final boolean binaryAsString;
+  private final boolean extractUnionsByType;
   private final ObjectFlattener<GenericRecord> avroFlattener;
   private final MapInputRowParser mapParser;
 
   @JsonCreator
   public AvroHadoopInputRowParser(
       @JsonProperty("parseSpec") ParseSpec parseSpec,
-      @JsonProperty("fromPigAvroStorage") Boolean fromPigAvroStorage
+      @JsonProperty("fromPigAvroStorage") Boolean fromPigAvroStorage,
+      @JsonProperty("binaryAsString") Boolean binaryAsString,
+      @JsonProperty("extractUnionsByType") Boolean extractUnionsByType
   )
   {
     this.parseSpec = parseSpec;
-    this.fromPigAvroStorage = fromPigAvroStorage == null ? false : fromPigAvroStorage;
-    this.avroFlattener = AvroParsers.makeFlattener(parseSpec, this.fromPigAvroStorage, false);
+    this.fromPigAvroStorage = fromPigAvroStorage != null && fromPigAvroStorage;
+    this.binaryAsString = binaryAsString != null && binaryAsString;
+    this.extractUnionsByType = extractUnionsByType != null && extractUnionsByType;
+    this.avroFlattener = AvroParsers.makeFlattener(parseSpec, this.fromPigAvroStorage, this.binaryAsString, this.extractUnionsByType);
     this.mapParser = new MapInputRowParser(parseSpec);
   }
 
@@ -71,6 +77,6 @@
   @Override
   public InputRowParser withParseSpec(ParseSpec parseSpec)
   {
-    return new AvroHadoopInputRowParser(parseSpec, fromPigAvroStorage);
+    return new AvroHadoopInputRowParser(parseSpec, fromPigAvroStorage, binaryAsString, extractUnionsByType);
   }
 }
diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java
index f375c63..fcb8bd8 100644
--- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java
+++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java
@@ -36,6 +36,8 @@
 public class AvroStreamInputRowParser implements ByteBufferInputRowParser
 {
   private final ParseSpec parseSpec;
+  private final Boolean binaryAsString;
+  private final Boolean extractUnionsByType;
   private final AvroBytesDecoder avroBytesDecoder;
   private final ObjectFlattener<GenericRecord> avroFlattener;
   private final MapInputRowParser mapParser;
@@ -43,12 +45,16 @@
   @JsonCreator
   public AvroStreamInputRowParser(
       @JsonProperty("parseSpec") ParseSpec parseSpec,
-      @JsonProperty("avroBytesDecoder") AvroBytesDecoder avroBytesDecoder
+      @JsonProperty("avroBytesDecoder") AvroBytesDecoder avroBytesDecoder,
+      @JsonProperty("binaryAsString") Boolean binaryAsString,
+      @JsonProperty("extractUnionsByType") Boolean extractUnionsByType
   )
   {
     this.parseSpec = Preconditions.checkNotNull(parseSpec, "parseSpec");
     this.avroBytesDecoder = Preconditions.checkNotNull(avroBytesDecoder, "avroBytesDecoder");
-    this.avroFlattener = AvroParsers.makeFlattener(parseSpec, false, false);
+    this.binaryAsString = binaryAsString != null && binaryAsString;
+    this.extractUnionsByType = extractUnionsByType != null && extractUnionsByType;
+    this.avroFlattener = AvroParsers.makeFlattener(parseSpec, false, this.binaryAsString, this.extractUnionsByType);
     this.mapParser = new MapInputRowParser(parseSpec);
   }
 
@@ -76,7 +82,9 @@
   {
     return new AvroStreamInputRowParser(
         parseSpec,
-        avroBytesDecoder
+        avroBytesDecoder,
+        binaryAsString,
+        extractUnionsByType
     );
   }
 
diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java
index 8f64532..8526a04 100644
--- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java
+++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java
@@ -44,13 +44,8 @@
 
 public class AvroFlattenerMaker implements ObjectFlatteners.FlattenerMaker<GenericRecord>
 {
-  private static final JsonProvider AVRO_JSON_PROVIDER = new GenericAvroJsonProvider();
-  private static final Configuration JSONPATH_CONFIGURATION =
-      Configuration.builder()
-                   .jsonProvider(AVRO_JSON_PROVIDER)
-                   .mappingProvider(new NotImplementedMappingProvider())
-                   .options(EnumSet.of(Option.SUPPRESS_EXCEPTIONS))
-                   .build();
+  private final JsonProvider avroJsonProvider;
+  private final Configuration jsonPathConfiguration;
 
   private static final EnumSet<Schema.Type> ROOT_TYPES = EnumSet.of(
       Schema.Type.STRING,
@@ -99,10 +94,18 @@
    * @param fromPigAvroStorage boolean to specify the data file is stored using AvroStorage
    * @param binaryAsString boolean to encode the byte[] as a string.
    */
-  public AvroFlattenerMaker(final boolean fromPigAvroStorage, final boolean binaryAsString)
+  public AvroFlattenerMaker(final boolean fromPigAvroStorage, final boolean binaryAsString, final boolean extractUnionsByType)
   {
     this.fromPigAvroStorage = fromPigAvroStorage;
     this.binaryAsString = binaryAsString;
+
+    this.avroJsonProvider = new GenericAvroJsonProvider(extractUnionsByType);
+    this.jsonPathConfiguration =
+        Configuration.builder()
+                     .jsonProvider(avroJsonProvider)
+                     .mappingProvider(new NotImplementedMappingProvider())
+                     .options(EnumSet.of(Option.SUPPRESS_EXCEPTIONS))
+                     .build();
   }
 
   @Override
@@ -126,7 +129,7 @@
   public Function<GenericRecord, Object> makeJsonPathExtractor(final String expr)
   {
     final JsonPath jsonPath = JsonPath.compile(expr);
-    return record -> transformValue(jsonPath.read(record, JSONPATH_CONFIGURATION));
+    return record -> transformValue(jsonPath.read(record, jsonPathConfiguration));
   }
 
   @Override
@@ -138,7 +141,7 @@
   @Override
   public JsonProvider getJsonProvider()
   {
-    return AVRO_JSON_PROVIDER;
+    return avroJsonProvider;
   }
 
   @Override
diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java
index 90c74f6..7417477 100644
--- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java
+++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java
@@ -42,6 +42,7 @@
   private static final Logger LOGGER = new Logger(AvroOCFInputFormat.class);
 
   private final boolean binaryAsString;
+  private final boolean extractUnionsByType;
   @Nullable
   private final Schema readerSchema;
 
@@ -50,7 +51,8 @@
       @JacksonInject @Json ObjectMapper mapper,
       @JsonProperty("flattenSpec") @Nullable JSONPathSpec flattenSpec,
       @JsonProperty("schema") @Nullable Map<String, Object> schema,
-      @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString
+      @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString,
+      @JsonProperty("extractUnionsByType") @Nullable Boolean extractUnionsByType
   ) throws Exception
   {
     super(flattenSpec);
@@ -62,7 +64,8 @@
     } else {
       this.readerSchema = null;
     }
-    this.binaryAsString = binaryAsString == null ? false : binaryAsString;
+    this.binaryAsString = binaryAsString != null && binaryAsString;
+    this.extractUnionsByType = extractUnionsByType != null && extractUnionsByType;
   }
 
   @Override
@@ -82,7 +85,8 @@
         temporaryDirectory,
         readerSchema,
         getFlattenSpec(),
-        binaryAsString
+        binaryAsString,
+        extractUnionsByType
     );
   }
 
diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java
index 97ab895..dc2a6d7 100644
--- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java
+++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java
@@ -56,14 +56,15 @@
       File temporaryDirectory,
       @Nullable Schema readerSchema,
       JSONPathSpec flattenSpec,
-      boolean binaryAsString
+      boolean binaryAsString,
+      boolean extractUnionsByType
   )
   {
     this.inputRowSchema = inputRowSchema;
     this.source = source;
     this.temporaryDirectory = temporaryDirectory;
     this.readerSchema = readerSchema;
-    this.recordFlattener = ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(false, binaryAsString));
+    this.recordFlattener = ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(false, binaryAsString, extractUnionsByType));
   }
 
   @Override
diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java
index 9d5665e..6d399a8 100644
--- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java
+++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java
@@ -39,7 +39,8 @@
   public static ObjectFlattener<GenericRecord> makeFlattener(
       final ParseSpec parseSpec,
       final boolean fromPigAvroStorage,
-      final boolean binaryAsString
+      final boolean binaryAsString,
+      final boolean extractUnionsByType
   )
   {
     final JSONPathSpec flattenSpec;
@@ -49,7 +50,7 @@
       flattenSpec = JSONPathSpec.DEFAULT;
     }
 
-    return ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(fromPigAvroStorage, binaryAsString));
+    return ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(fromPigAvroStorage, binaryAsString, extractUnionsByType));
   }
 
   public static List<InputRow> parseGenericRecord(
diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java
index 3b59b36..d42f359 100644
--- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java
+++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java
@@ -35,6 +35,7 @@
 public class AvroStreamInputFormat extends NestedInputFormat
 {
   private final boolean binaryAsString;
+  private final boolean extractUnionsByType;
 
   private final AvroBytesDecoder avroBytesDecoder;
 
@@ -42,12 +43,14 @@
   public AvroStreamInputFormat(
       @JsonProperty("flattenSpec") @Nullable JSONPathSpec flattenSpec,
       @JsonProperty("avroBytesDecoder") AvroBytesDecoder avroBytesDecoder,
-      @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString
+      @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString,
+      @JsonProperty("extractUnionsByType") @Nullable Boolean extractUnionsByType
   )
   {
     super(flattenSpec);
     this.avroBytesDecoder = avroBytesDecoder;
-    this.binaryAsString = binaryAsString == null ? false : binaryAsString;
+    this.binaryAsString = binaryAsString != null && binaryAsString;
+    this.extractUnionsByType = extractUnionsByType != null && extractUnionsByType;
   }
 
   @Override
@@ -76,7 +79,8 @@
         source,
         avroBytesDecoder,
         getFlattenSpec(),
-        binaryAsString
+        binaryAsString,
+        extractUnionsByType
     );
   }
 
diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java
index d8d92eb..c04ec14 100644
--- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java
+++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java
@@ -52,13 +52,14 @@
       InputEntity source,
       AvroBytesDecoder avroBytesDecoder,
       JSONPathSpec flattenSpec,
-      boolean binaryAsString
+      boolean binaryAsString,
+      boolean extractUnionsByType
   )
   {
     this.inputRowSchema = inputRowSchema;
     this.source = source;
     this.avroBytesDecoder = avroBytesDecoder;
-    this.recordFlattener = ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(false, binaryAsString));
+    this.recordFlattener = ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(false, binaryAsString, extractUnionsByType));
   }
 
   @Override
diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java
index ab6a53e..86bbb7d 100644
--- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java
+++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java
@@ -19,15 +19,19 @@
 
 package org.apache.druid.data.input.avro;
 
+import com.google.common.collect.ImmutableMap;
 import com.jayway.jsonpath.InvalidJsonException;
 import com.jayway.jsonpath.spi.json.JsonProvider;
 import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericEnumSymbol;
+import org.apache.avro.generic.GenericFixed;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.util.Utf8;
 
 import javax.annotation.Nullable;
 
 import java.io.InputStream;
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -41,6 +45,13 @@
  */
 public class GenericAvroJsonProvider implements JsonProvider
 {
+  private final boolean extractUnionsByType;
+
+  GenericAvroJsonProvider(final boolean extractUnionsByType)
+  {
+    this.extractUnionsByType = extractUnionsByType;
+  }
+
   @Override
   public Object parse(final String s) throws InvalidJsonException
   {
@@ -148,7 +159,11 @@
     if (o == null) {
       return null;
     } else if (o instanceof GenericRecord) {
-      return ((GenericRecord) o).get(s);
+      final GenericRecord record = (GenericRecord) o;
+      if (extractUnionsByType && isExtractableUnion(record.getSchema().getField(s))) {
+        return extractUnionTypes(record.get(s));
+      }
+      return record.get(s);
     } else if (o instanceof Map) {
       final Map theMap = (Map) o;
       if (theMap.containsKey(s)) {
@@ -195,4 +210,46 @@
   {
     return o;
   }
+
+  private boolean isExtractableUnion(final Schema.Field field)
+  {
+    return field.schema().isUnion() &&
+           field.schema().getTypes().stream().filter(type -> type.getType() != Schema.Type.NULL).count() > 1;
+  }
+
+  private Map<String, Object> extractUnionTypes(final Object o)
+  {
+    // Primitive types and unnamped complex types are keyed their type name.
+    // Complex named types are keyed by their names.
+    // This is safe because an Avro union can only contain a single member of each unnamed type and duplicates
+    // of the same named type are not allowed. i.e only a single array is allowed, multiple records are allowed as
+    // long as each has a unique name.
+    // The Avro null type is elided as it's value can only ever be null
+    if (o instanceof Integer) {
+      return ImmutableMap.of("int", o);
+    } else if (o instanceof Long) {
+      return ImmutableMap.of("long", o);
+    } else if (o instanceof Float) {
+      return ImmutableMap.of("float", o);
+    } else if (o instanceof Double) {
+      return ImmutableMap.of("double", o);
+    } else if (o instanceof Boolean) {
+      return ImmutableMap.of("boolean", o);
+    } else if (o instanceof Utf8) {
+      return ImmutableMap.of("string", o);
+    } else if (o instanceof ByteBuffer) {
+      return ImmutableMap.of("bytes", o);
+    } else if (o instanceof Map) {
+      return ImmutableMap.of("map", o);
+    } else if (o instanceof List) {
+      return ImmutableMap.of("array", o);
+    } else if (o instanceof GenericRecord) {
+      return ImmutableMap.of(((GenericRecord) o).getSchema().getName(), o);
+    } else if (o instanceof GenericFixed) {
+      return ImmutableMap.of(((GenericFixed) o).getSchema().getName(), o);
+    } else if (o instanceof GenericEnumSymbol) {
+      return ImmutableMap.of(((GenericEnumSymbol<?>) o).getSchema().getName(), o);
+    }
+    return ImmutableMap.of();
+  }
 }
diff --git a/extensions-core/avro-extensions/src/test/avro/some-datum.avsc b/extensions-core/avro-extensions/src/test/avro/some-datum.avsc
index bf8ea68..f03cb70 100644
--- a/extensions-core/avro-extensions/src/test/avro/some-datum.avsc
+++ b/extensions-core/avro-extensions/src/test/avro/some-datum.avsc
@@ -1,38 +1,181 @@
-[{
-  "namespace": "org.apache.druid.data.input",
-	"name": "SomeAvroDatum",
-	"type": "record",
-	"fields" : [
-		{"name":"timestamp","type":"long"},
-		{"name":"eventType","type":"string"},
-		{"name":"id","type":"long"},
-		{"name":"someOtherId","type":"long"},
-		{"name":"isValid","type":"boolean"},
-		{"name":"someIntArray","type":{"type":"array","items":"int"}},
-		{"name":"someStringArray","type":{"type":"array","items":"string"}},
-		{"name":"someIntValueMap","type":{"type":"map","values":"int"}},
-		{"name":"someStringValueMap","type":{"type":"map","values":"string"}},
-		{"name":"someUnion","type":["null","string"]},
-		{"name":"someNull","type":"null"},
-    {"name":"someFixed","type":{"type":"fixed","size":16,"name":"MyFixed"}},
-    {"name":"someBytes","type":"bytes"},
-    {"name":"someEnum","type":{"type":"enum","name":"MyEnum","symbols":["ENUM0","ENUM1","ENUM2"]}},
-    {"name":"someRecord","type":{
-      "type":"record","name":"MySubRecord","fields":[
-        {"name":"subInt","type":"int"},
-        {"name":"subLong","type":"long"}
-      ]
-    }},
-
-   	{"name":"someLong","type":"long"},
-		{"name":"someInt","type":"int"},
-		{"name":"someFloat","type":"float"},
-		{"name":"someRecordArray","type":{
-		  "type":"array","items":{
-		   "type":"record","name":"MyNestedRecord","fields":[
-		    {"name":"nestedString","type":"string"}
+[
+  {
+    "namespace": "org.apache.druid.data.input",
+    "name": "SomeAvroDatum",
+    "type": "record",
+    "fields": [
+      {
+        "name": "timestamp",
+        "type": "long"
+      },
+      {
+        "name": "eventType",
+        "type": "string"
+      },
+      {
+        "name": "id",
+        "type": "long"
+      },
+      {
+        "name": "someOtherId",
+        "type": "long"
+      },
+      {
+        "name": "isValid",
+        "type": "boolean"
+      },
+      {
+        "name": "someIntArray",
+        "type": {
+          "type": "array",
+          "items": "int"
+        }
+      },
+      {
+        "name": "someStringArray",
+        "type": {
+          "type": "array",
+          "items": "string"
+        }
+      },
+      {
+        "name": "someIntValueMap",
+        "type": {
+          "type": "map",
+          "values": "int"
+        }
+      },
+      {
+        "name": "someStringValueMap",
+        "type": {
+          "type": "map",
+          "values": "string"
+        }
+      },
+      {
+        "name": "someUnion",
+        "type": [
+          "null",
+          "string"
         ]
-       }
-     }}
-	]
-}]
+      },
+      {
+        "name": "someMultiMemberUnion",
+        "type": [
+          "int",
+          "long",
+          "float",
+          "double",
+          "boolean",
+          "string",
+          "bytes",
+          {
+            "type": "map",
+            "values": "string"
+          },
+          {
+            "type": "array",
+            "items": "int"
+          },
+          {
+            "type": "record",
+            "name": "UnionSubRecord",
+            "fields": [
+              {
+                "name": "subString",
+                "type": "string"
+              }
+            ]
+          },
+          {
+            "name": "UnionSubFixed",
+            "type": "fixed",
+            "size": 10
+          },
+          {
+            "type": "enum",
+            "name": "UnionSubEnum",
+            "symbols": [
+              "ENUM0",
+              "ENUM1",
+              "ENUM2"
+            ]
+          }
+        ]
+      },
+      {
+        "name": "someNull",
+        "type": "null"
+      },
+      {
+        "name": "someFixed",
+        "type": {
+          "type": "fixed",
+          "size": 16,
+          "name": "MyFixed"
+        }
+      },
+      {
+        "name": "someBytes",
+        "type": "bytes"
+      },
+      {
+        "name": "someEnum",
+        "type": {
+          "type": "enum",
+          "name": "MyEnum",
+          "symbols": [
+            "ENUM0",
+            "ENUM1",
+            "ENUM2"
+          ]
+        }
+      },
+      {
+        "name": "someRecord",
+        "type": {
+          "type": "record",
+          "name": "MySubRecord",
+          "fields": [
+            {
+              "name": "subInt",
+              "type": "int"
+            },
+            {
+              "name": "subLong",
+              "type": "long"
+            }
+          ]
+        }
+      },
+      {
+        "name": "someLong",
+        "type": "long"
+      },
+      {
+        "name": "someInt",
+        "type": "int"
+      },
+      {
+        "name": "someFloat",
+        "type": "float"
+      },
+      {
+        "name": "someRecordArray",
+        "type": {
+          "type": "array",
+          "items": {
+            "type": "record",
+            "name": "MyNestedRecord",
+            "fields": [
+              {
+                "name": "nestedString",
+                "type": "string"
+              }
+            ]
+          }
+        }
+      }
+    ]
+  }
+]
diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroHadoopInputRowParserTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroHadoopInputRowParserTest.java
index 6e3cdb0..c6ade85 100644
--- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroHadoopInputRowParserTest.java
+++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroHadoopInputRowParserTest.java
@@ -62,7 +62,7 @@
 
   private void testParse(GenericRecord record, boolean fromPigAvroStorage) throws IOException
   {
-    AvroHadoopInputRowParser parser = new AvroHadoopInputRowParser(AvroStreamInputRowParserTest.PARSE_SPEC, fromPigAvroStorage);
+    AvroHadoopInputRowParser parser = new AvroHadoopInputRowParser(AvroStreamInputRowParserTest.PARSE_SPEC, fromPigAvroStorage, false, false);
     AvroHadoopInputRowParser parser2 = jsonMapper.readValue(
         jsonMapper.writeValueAsBytes(parser),
         AvroHadoopInputRowParser.class
diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputFormatTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputFormatTest.java
index 139616b..918aa0a 100644
--- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputFormatTest.java
+++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputFormatTest.java
@@ -118,6 +118,7 @@
     AvroStreamInputFormat inputFormat = new AvroStreamInputFormat(
         flattenSpec,
         new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository),
+        false,
         false
     );
     NestedInputFormat inputFormat2 = jsonMapper.readValue(
@@ -134,6 +135,7 @@
     AvroStreamInputFormat inputFormat = new AvroStreamInputFormat(
         flattenSpec,
         new SchemaRegistryBasedAvroBytesDecoder("http://test:8081", 100, null, null, null),
+        false,
         false
     );
     NestedInputFormat inputFormat2 = jsonMapper.readValue(
@@ -150,6 +152,7 @@
     AvroStreamInputFormat inputFormat = new AvroStreamInputFormat(
         flattenSpec,
         new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository),
+        false,
         false
     );
     NestedInputFormat inputFormat2 = jsonMapper.readValue(
@@ -193,6 +196,7 @@
     AvroStreamInputFormat inputFormat = new AvroStreamInputFormat(
         flattenSpec,
         new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository),
+        false,
         false
     );
     NestedInputFormat inputFormat2 = jsonMapper.readValue(
diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java
index 0dbb6b9..8ed9f44 100644
--- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java
+++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java
@@ -151,6 +151,7 @@
       }
   );
   private static final String SOME_UNION_VALUE = "string as union";
+  private static final Integer SOME_UNION_MEMBER_VALUE = 1;
   private static final ByteBuffer SOME_BYTES_VALUE = ByteBuffer.allocate(8);
   private static final String SOME_RECORD_STRING_VALUE = "string in record";
   private static final List<MyNestedRecord> SOME_RECORD_ARRAY_VALUE = Collections.singletonList(MyNestedRecord.newBuilder()
@@ -176,7 +177,9 @@
     Repository repository = new Avro1124RESTRepositoryClientWrapper("http://github.io");
     AvroStreamInputRowParser parser = new AvroStreamInputRowParser(
         PARSE_SPEC,
-        new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository)
+        new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository),
+        false,
+        false
     );
     ByteBufferInputRowParser parser2 = jsonMapper.readValue(
         jsonMapper.writeValueAsString(parser),
@@ -193,7 +196,9 @@
     Repository repository = new InMemoryRepository(null);
     AvroStreamInputRowParser parser = new AvroStreamInputRowParser(
         PARSE_SPEC,
-        new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository)
+        new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository),
+        false,
+        false
     );
     ByteBufferInputRowParser parser2 = jsonMapper.readValue(
         jsonMapper.writeValueAsString(parser),
@@ -234,7 +239,9 @@
     Repository repository = new InMemoryRepository(null);
     AvroStreamInputRowParser parser = new AvroStreamInputRowParser(
         PARSE_SPEC_SCHEMALESS,
-        new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository)
+        new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository),
+        false,
+        false
     );
     ByteBufferInputRowParser parser2 = jsonMapper.readValue(
         jsonMapper.writeValueAsString(parser),
@@ -354,7 +361,7 @@
     Assert.assertEquals(SOME_INT_VALUE, inputRow.getMetric("someInt"));
   }
 
-  public static SomeAvroDatum buildSomeAvroDatum()
+  private static SomeAvroDatum.Builder createSomeAvroDatumBuilderDefaults()
   {
     return SomeAvroDatum.newBuilder()
                         .setTimestamp(DATE_TIME.toInstant().toEpochMilli())
@@ -370,12 +377,24 @@
                         .setSomeIntValueMap(SOME_INT_VALUE_MAP_VALUE)
                         .setSomeStringValueMap(SOME_STRING_VALUE_MAP_VALUE)
                         .setSomeUnion(SOME_UNION_VALUE)
+                        .setSomeMultiMemberUnion(SOME_UNION_MEMBER_VALUE)
                         .setSomeFixed(SOME_FIXED_VALUE)
                         .setSomeBytes(SOME_BYTES_VALUE)
                         .setSomeNull(null)
                         .setSomeEnum(MyEnum.ENUM1)
                         .setSomeRecord(SOME_RECORD_VALUE)
-                        .setSomeRecordArray(SOME_RECORD_ARRAY_VALUE)
-                        .build();
+                        .setSomeRecordArray(SOME_RECORD_ARRAY_VALUE);
+  }
+
+  public static SomeAvroDatum buildSomeAvroDatum()
+  {
+    return createSomeAvroDatumBuilderDefaults().build();
+  }
+
+  public static SomeAvroDatum buildSomeAvroDatumWithUnionValue(Object unionValue)
+  {
+    return createSomeAvroDatumBuilderDefaults()
+        .setSomeMultiMemberUnion(unionValue)
+        .build();
   }
 }
diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java
index 0761672..e064e7a 100644
--- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java
+++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java
@@ -19,11 +19,19 @@
 
 package org.apache.druid.data.input.avro;
 
+import org.apache.avro.util.Utf8;
 import org.apache.druid.data.input.AvroStreamInputRowParserTest;
 import org.apache.druid.data.input.SomeAvroDatum;
+import org.apache.druid.data.input.UnionSubEnum;
+import org.apache.druid.data.input.UnionSubFixed;
+import org.apache.druid.data.input.UnionSubRecord;
 import org.junit.Assert;
 import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 
 public class AvroFlattenerMakerTest
@@ -33,7 +41,7 @@
   public void getRootField()
   {
     final SomeAvroDatum record = AvroStreamInputRowParserTest.buildSomeAvroDatum();
-    final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false);
+    final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false, true);
 
     Assert.assertEquals(
         record.getTimestamp(),
@@ -120,7 +128,7 @@
   public void makeJsonPathExtractor()
   {
     final SomeAvroDatum record = AvroStreamInputRowParserTest.buildSomeAvroDatum();
-    final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false);
+    final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false, true);
 
     Assert.assertEquals(
         record.getTimestamp(),
@@ -163,6 +171,10 @@
         flattener.makeJsonPathExtractor("$.someUnion").apply(record)
     );
     Assert.assertEquals(
+        record.getSomeMultiMemberUnion(),
+        flattener.makeJsonPathExtractor("$.someMultiMemberUnion.int").apply(record)
+    );
+    Assert.assertEquals(
         record.getSomeNull(),
         flattener.makeJsonPathExtractor("$.someNull").apply(record)
     );
@@ -219,11 +231,75 @@
     );
   }
 
+  @Test
+  public void jsonPathExtractorExtractUnionsByType()
+  {
+    final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false, true);
+
+    // Unmamed types are accessed by type
+
+    // int
+    Assert.assertEquals(1, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.int").apply(
+        AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(1)));
+
+    // long
+    Assert.assertEquals(1L, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.long").apply(
+        AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(1L)));
+
+    // float
+    Assert.assertEquals((float) 1.0, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.float").apply(
+        AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue((float) 1.0)));
+
+    // double
+    Assert.assertEquals(1.0, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.double").apply(
+        AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(1.0)));
+
+    // string
+    Assert.assertEquals("string", flattener.makeJsonPathExtractor("$.someMultiMemberUnion.string").apply(
+        AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(new Utf8("string"))));
+
+    // bytes
+    Assert.assertArrayEquals(new byte[] {1}, (byte[]) flattener.makeJsonPathExtractor("$.someMultiMemberUnion.bytes").apply(
+        AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(ByteBuffer.wrap(new byte[] {1}))));
+
+    // map
+    Assert.assertEquals(2, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.map.two").apply(
+        AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(new HashMap<String, Integer>() {{
+            put("one", 1);
+            put("two", 2);
+            put("three", 3);
+          }
+        }
+        )));
+
+    // array
+    Assert.assertEquals(3, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.array[2]").apply(
+        AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(Arrays.asList(1, 2, 3))));
+
+    // Named types are accessed by name
+
+    // record
+    Assert.assertEquals("subRecordString", flattener.makeJsonPathExtractor("$.someMultiMemberUnion.UnionSubRecord.subString").apply(
+        AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(
+            UnionSubRecord.newBuilder()
+                          .setSubString("subRecordString")
+                          .build())));
+
+    // fixed
+    final byte[] fixedBytes = new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
+    Assert.assertEquals(fixedBytes, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.UnionSubFixed").apply(
+        AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(new UnionSubFixed(fixedBytes))));
+
+    // enum
+    Assert.assertEquals(String.valueOf(UnionSubEnum.ENUM1), flattener.makeJsonPathExtractor("$.someMultiMemberUnion.UnionSubEnum").apply(
+        AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(UnionSubEnum.ENUM1)));
+  }
+
   @Test(expected = UnsupportedOperationException.class)
   public void makeJsonQueryExtractor()
   {
     final SomeAvroDatum record = AvroStreamInputRowParserTest.buildSomeAvroDatum();
-    final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false);
+    final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false, false);
 
     Assert.assertEquals(
         record.getTimestamp(),
diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java
index 4841483..b00c1e4 100644
--- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java
+++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java
@@ -144,7 +144,7 @@
       Assert.assertFalse(iterator.hasNext());
       final Map<String, Object> rawColumns = row.getRawValues();
       Assert.assertNotNull(rawColumns);
-      Assert.assertEquals(19, rawColumns.size());
+      Assert.assertEquals(20, rawColumns.size());
       final List<InputRow> inputRows = row.getInputRows();
       Assert.assertNotNull(inputRows);
       final InputRow inputRow = Iterables.getOnlyElement(inputRows);
@@ -201,7 +201,7 @@
     final DimensionsSpec dimensionsSpec = new DimensionsSpec(DimensionsSpec.getDefaultSchemas(ImmutableList.of(
         "eventType")));
 
-    final AvroOCFInputFormat inputFormat = new AvroOCFInputFormat(mapper, null, readerSchema, null);
+    final AvroOCFInputFormat inputFormat = new AvroOCFInputFormat(mapper, null, readerSchema, null, null);
     final InputRowSchema schema = new InputRowSchema(timestampSpec, dimensionsSpec, ColumnsFilter.all());
     final FileEntity entity = new FileEntity(someAvroFile);
     return inputFormat.createReader(schema, entity, temporaryFolder.newFolder());
diff --git a/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java b/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java
index c8e9a45..796891e 100755
--- a/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java
+++ b/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java
@@ -51,6 +51,7 @@
 {
   private final ParseSpec parseSpec;
   private final boolean binaryAsString;
+  private final boolean extractUnionsByType;
   private final TimestampSpec timestampSpec;
   private final ObjectFlattener<GenericRecord> recordFlattener;
   private final List<String> dimensions;
@@ -58,13 +59,15 @@
   @JsonCreator
   public ParquetAvroHadoopInputRowParser(
       @JsonProperty("parseSpec") ParseSpec parseSpec,
-      @JsonProperty("binaryAsString") Boolean binaryAsString
+      @JsonProperty("binaryAsString") Boolean binaryAsString,
+      @JsonProperty("extractUnionsByType") Boolean extractUnionsByType
   )
   {
     this.parseSpec = parseSpec;
     this.timestampSpec = parseSpec.getTimestampSpec();
     this.dimensions = parseSpec.getDimensionsSpec().getDimensionNames();
-    this.binaryAsString = binaryAsString == null ? false : binaryAsString;
+    this.binaryAsString = binaryAsString != null && binaryAsString;
+    this.extractUnionsByType = extractUnionsByType != null && extractUnionsByType;
 
     final JSONPathSpec flattenSpec;
     if (parseSpec instanceof AvroParseSpec) {
@@ -75,7 +78,7 @@
 
     this.recordFlattener = ObjectFlatteners.create(
         flattenSpec,
-        new AvroFlattenerMaker(false, this.binaryAsString)
+        new AvroFlattenerMaker(false, this.binaryAsString, this.extractUnionsByType)
     );
   }
 
@@ -131,6 +134,6 @@
   @Override
   public InputRowParser withParseSpec(ParseSpec parseSpec)
   {
-    return new ParquetAvroHadoopInputRowParser(parseSpec, binaryAsString);
+    return new ParquetAvroHadoopInputRowParser(parseSpec, binaryAsString, extractUnionsByType);
   }
 }
diff --git a/website/.spelling b/website/.spelling
index dd240c5..a0a1025 100644
--- a/website/.spelling
+++ b/website/.spelling
@@ -74,6 +74,7 @@
 EMRFS
 ETL
 Elasticsearch
+Enums
 FirehoseFactory
 FlattenSpec
 Float.NEGATIVE_INFINITY