Add support for Iceberg MR / InputFormat and OutputFormat APIs

[rdsr]

It would be great if we added support for InputFormat and OutputFormat APIs. There's already an InputFormat written but it is specific to Pig and under the iceberg-pig module. @rdblue what's your take if we make the necessary changes in that InputFormat and maybe implement an OutputFormat and have these as part of an iceberg-mr module ?

[rdblue]

I'd be up for reviewing this.

[guilload]

Hey,

I started working on an MR InputFormat for Iceberg since there’s some overlap with the InputFormat I’m working on for Hive. I have a few questions:

– Thoughts on using GenericRecord as data container for both the Input and Output formats?

– IcebergPigInputFormat declares a public constructor IcebergPigInputFormat(Table table) but I don’t think that’s an option for MR since the JobSubmitter class instantiates the input format via reflection without constructor arguments. How do we solve this? Have the user pass the canonical name of the Catalog class they want to use along with a table identifier in the job config and then instantiate the catalog via reflection and finally load the table?

– Regarding serialization, IcebergPigInputFormat relies on Java serialization via org.apache.pig.impl.util.ObjectSerializer, shall I reuse the same approach or use a different serialization mechanism?

Thanks!

[rdblue]

IcebergPigInputFormat relies on Java serialization via org.apache.pig.impl.util.ObjectSerializer, shall I reuse the same approach or use a different serialization mechanism?

I don't think a generic InputFormat should rely on Pig, so using Java serialization where necessary is a better choice.

Also, there are some situations where you'd want to use other options:

• Classes that have JSON parsers, like Schema and PartitionSpec should serialize to/from JSON because that produces a human-readable string
• Classes with a serialization requirement from MR should use that. For example, InputSplit needs to be Writeable, so I would expect an IcebergInputSplit to correctly serialize itself using the Writeable interface.

Thoughts on using GenericRecord as data container for both the Input and Output formats?

I'd use Iceberg's Record interface instead. GenericRecord is the implementation class. It's also the name of an Avro class, which we shouldn't expose.

IcebergPigInputFormat declares a public constructor IcebergPigInputFormat(Table table) but I don’t think that’s an option for MR since the JobSubmitter class instantiates the input format via reflection without constructor arguments. How do we solve this?

The easiest solution is to use HiveCatalogs like we do in Spark and set the configuration on the catalog that gets returned. The drawback there is that we didn't intend to use HiveCatalogs for very long because Spark has a catalog plugin system coming in 3.0 to handle this.

Another option is instantiating using a no-arg constructor and setting Configuration if the Catalog implements Configurable. That sounds reasonable to me, but is not very generic.

Last, you could add a ServiceLoader that acts as a factory for catalogs. Then you could load a service by name and call getCatalog(Configuration) on it.

[rdsr]

All good points.

I'd use Iceberg's Record interface instead. GenericRecord is the implementation class. It's also the name of an Avro class, which we shouldn't expose

For this I think it should be possible to parameterize [using generics] the container and provide a default implementation for Iceberg's Record type. Users may want to directly read Avro data or in their own container, they shouldn't have to re-convert from Iceberg's Record to, say, Avro.

[guilload]

Ok, thanks for the pointers. That's helpful. I'll start with HiveCatalogs and we can revisit later.

One last question regarding generic container. Let's assume a user wants to user a custom container: FooRecord. For each file format, the user will have to provide a "value reader function", for instance a Schema -> OrcValueReader<FooRecord> function and a MessageType -> ParquetValueReader<FooRecord> function.

What the interface for providing those custom value reader functions should look like? For instance, parquet-mr relies on the ReadSupport interface to solve a similar problem and ships with a default GroupReadSupport implementation.

Here's my WIP branch:
guilload#1

[rdblue]

@guilload, good points about the reader functions. This is an area where we haven't invested more time because we are currently only supporting a small set of in-memory representations. What I would like is to eventually make a way to plug in representations for struct, map, and list and have a generic way to support those across data formats.

Consider struct, for example. Iceberg uses an interface, StructLike, to work with struct objects. So if we write a ValueReader for Parquet and a ValueReader for Avro that supports a StructLike, we could then add an interface like this for plugging in a custom type:

interface DataModel<S extends StructLike, M extends Map, L extends List> {
 S newStruct();
 M newMap();
 L newList();
}

Some of the value readers are already generic and can also convert from an intermediate representation to a final representation. If we supported intermediate and final representations, then your final type wouldn't even need to implement StructLike. You could use a wrapper to translate from StructLike to your record's API and then return your record.

[teabot]

@guilload Has this work progressed any further since 8 Aug 2019? I ask because we are considering tackling the integration with Hive and do not wish to duplicate efforts.

[rdsr]

@guilload, do you plan to work on this? We are planning to do some work for Hive read support using these MR apis. Let us know if you need any help with this, or we can even build on top of your work.

[rdsr]

fyi, I'm taking this patch up and working on an initial version using @guilload patch as base

[massdosage]

Hey there, along with @teabot and @cmathiesen we've started working on this and have managed to get the read path working from Hive on an Iceberg table using an InputFormat. We had to make quite a few changes to the original patch as it was using a different Hive API to that one that appears to be in use in Hive 2.x. Maybe it was targetting Hive 3.x? We have run into a number of issues with version conflicts between Hive, Spark, Iceberg, Guava and Avro but managed to get it working as a separate (Maven) project. We attempted to integrate it with Iceberg as a new module (although maybe adding it to the existing iceberg-hive would be better) but couldn't get Gradle to isolate the versions for just the new subproject. We'd be happy to share what we've done so far, we just need to figure out where best to put it so others can take a look.

[rdsr]

@massdosage that's great to hear.

We had to make quite a few changes to the original patch as it was using a different Hive API

Which patch is this ? @guilload patch was only for MR input format AFAIK.

My take would be that the MR API patches should go into the iceberg-mr module. The Hive StorageHandler API can possibly go into iceberg-hive, but I guess you may be using a different version of Hive then Iceberg. This can be a non-issue if the StorageHandler API has not changed across different versions of Hive.
Nevertheless, it would be great if you can have the patch out. We can review it and take it from there.
Thanks again all of you for your work on this!

[massdosage]

Yes, the InputFormat in the other patch was using org.apache.hadoop.mapreduce.InputFormat; but in order to get this working with Hive 2.3.x we had to use org.apache.hadoop.mapred.InputFormat. We can get the code up onto GitHub to show and then you could pull the parts you want into Iceberg if it looks useful. We'll try get that done in a day or two.

Our other attempt at it was in an iceberg-mr subproject but ended in version lock hell, it's quite possible that there are some Gradle tricks to manage versions that we're not aware of as we're not big Gradle users. The "work in progress" code for that is here: https://github.com/ExpediaGroup/incubator-iceberg/blob/hive-read-support/mr/src/main/java/org/apache/iceberg/mr/IcebergInputFormat.java. We also are aware that there are probably way better ways to get to the Iceberg metadata instead of the path/string manipulation we are currently doing but we're not entirely aware of all the ins and outs of the Iceberg APIs.

[rdblue]

Is an update to Hive 2.3 the main blocker for getting this into the Apache repo? If so, maybe we should look at upgrading Hive earlier.

[colinmjj]

hi, @rdsr @massdosage @rdblue
I also did some investigation on Hive integrate with Iceberg, the target of investigation is to integrate Hive with possible plugin,eg, HiveStorageHandler, but got some block issues:
file/folder of data has different organization, in Hive, there is:

• warehouse
  • dbname.db
    • table
      • partition
        • datafile

in Iceberg, it maybe:

• catalog_base
  • metadata
    • manifest
  • datafile

It's wired to have both folder in warehouse, so I suppose iceberg table may be existed as Hive external table, and define the metadata folder as table location which will be used in InputFormat.

Partition is also a problem. Partition is existed as a folder in Hive, but it's "hidden partition" in Iceberg. There is no plugin like InputFormat to solve this, and iceberg can't work properly now. For example, "show partition", "select c1 from t where partition_col = 'p1'".
I think Hive has to be modified to deal with the partition things, eg, "alter table add partition", "show partition", etc.

Our scene is writing Iceberg with Flink and Hive can read the data ASAP, as an option solution, maybe create a toolkit which can transfer IcebergTable to Hive table is suitable for our case.
Do we have the same target on Hive integration in this MR?

[teabot]

Hey Colin, In the implementation described by @massdosage, we assume that Iceberg is responsible for partitioning, layout of files, etc. This allows us to model Iceberg tables as unpartitioned external Hive tables in an existing metastore. By reducing scope like this, we avoid the need to align the Hive partitioning and warehouse models with that of Iceberg. Additionally, we can have Iceberg tables coexist with existing Hive tables in the same catalogue - which is important for us as any move towards Iceberg will be an incremental migration. Thanks, Elliot.

…

On Wed, 4 Mar 2020 at 09:07, Colin ***@***.***> wrote: hi, @rdsr <https://github.com/rdsr> @massdosage <https://github.com/massdosage> @rdblue <https://github.com/rdblue> I also did some investigation on Hive integrate with Iceberg, the target of investigation is to integrate Hive with possible plugin,eg, HiveStorageHandler, but got some block issues: 1. file/folder of data has different organization, in Hive, there is: warehouse |- dbname.db |- table |- partition |- datafile in Iceberg, it maybe: catalog_base |- metadata |- manifest |- datafile It's wired to have both folder in warehouse, so I suppose iceberg table may be existed as Hive external table, and define the metadata folder as table location which will be used in InputFormat. 2. Partition is also a problem. Partition is existed as a folder in Hive, but it's "hidden partition" in Iceberg. There is no plugin like InputFormat to solve this, and iceberg can't work properly now. For example, "show partition", "select c1 from t where partition_col = 'p1'". I think Hive has to be modified to deal with the partition things, eg, "alter table add partition", "show partition", etc. Our scene is writing Iceberg with Flink and Hive can read the data ASAP, as an option solution, maybe create a toolkit which can transfer IcebergTable to Hive table is suitable for our case. Do we have the same target on Hive integration in this MR? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#170>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AABX4VVPNNU5M5OZ7K5IMHLRFYK6NANCNFSM4HH46HTA> .