fix tfrecord compress on hdfs by bigbear2017 · Pull Request #108 · tensorflow/ecosystem

bigbear2017 · 2018-10-30T03:50:41Z

Default code doesn't support gzip or other compression type on hdfs. This
has been fixed in this pull request. Simply using following gzip compression,
it can reduce 6-7 times less space.

    TFRecordFileOutputFormat.setCompressOutput(job, true);
    TFRecordFileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);

googlebot · 2018-10-30T03:50:43Z

Thanks for your pull request. It looks like this may be your first contribution to a Google open source project (if not, look below for help). Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

📝 Please visit https://cla.developers.google.com/ to sign.

Once you've signed (or fixed any issues), please reply here (e.g. I signed it!) and we'll verify it.

What to do if you already signed the CLA

Individual signers

It's possible we don't have your GitHub username or you're using a different email address on your commit. Check your existing CLA data and verify that your email is set on your git commits.

Corporate signers

Your company has a Point of Contact who decides which employees are authorized to participate. Ask your POC to be added to the group of authorized contributors. If you don't know who your Point of Contact is, direct the Google project maintainer to go/cla#troubleshoot (Public version).
The email used to register you as an authorized contributor must be the email used for the Git commit. Check your existing CLA data and verify that your email is set on your git commits.
The email used to register you as an authorized contributor must also be attached to your GitHub account.

bigbear2017 · 2018-10-30T03:53:56Z

I signed it!

googlebot · 2018-10-30T03:53:57Z

CLAs look good, thanks!

jhseu · 2018-11-13T23:01:10Z

The underlying TFRecords file format supports compression, and I'm worried that the difference in format here will confuse users. Could we use that instead?

bigbear2017 · 2018-11-30T06:26:02Z

The underlying TFRecords file format supports compression, and I'm worried that the difference in format here will confuse users. Could we use that instead?

please refer to this issue:
#61

Now, the compression doesn't work on spark or hadoop MR. If you write data local, it's OK. You can compress TF records. The compression is done by python not TF. To me, there is no confusion. With this PR, you can read compressed data generated by Spark or hadoop in python. This has been tested in our system.

sahilbadyal · 2019-03-08T12:58:08Z

@bigbear2017 Do we have to add options to compress the TFRecord files ?

bigbear2017 · 2019-03-10T07:17:32Z

@sahilbadyal
Just add the following code to your job.
TFRecordFileOutputFormat.setCompressOutput(job, true);
TFRecordFileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);

sahilbadyal · 2019-03-13T08:07:43Z

@bigbear2017 Sorry for not asking the question right. I am using Pyspark

fhoering · 2019-03-14T12:19:24Z

Why is this this PR closed ? It has not been merged.
We have exactly the same issue. We need to write compressed TFRecords from Spark to HDFS. Compressing them manually is yet another step to do. In Python TFRecordWriter supports this with TFRecordCompressionType
Why not reusing the settings from FileOutputFormat as proposed here ? As it uses the Hadoop RecordWriter API it seems the standard way to do.

fhoering · 2019-03-14T14:35:43Z

@@ -32,20 +32,29 @@ public class TFRecordFileOutputFormat extends FileOutputFormat<BytesWritable, Nu
  @Override public RecordWriter<BytesWritable, NullWritable> getRecordWriter(


relevant imports are missing:
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.util.ReflectionUtils;

import java.io.DataOutputStream;

fhoering · 2019-03-14T14:37:04Z

+    FileSystem fs = file.getFileSystem(conf);
+    final DataOutputStream fsdos = isCompressed ?
+      new DataOutputStream(codec.createOutputStream(fs.create(file, true, bufferSize))) :
+      fs.create(file, true, bufferSize);


fs.create(file, true, bufferSize) could be factorized in a variable

fhoering · 2019-03-14T14:59:48Z

I tested the fix. It is working (with minor changes, need to add the right imports) and can be imported from Python by setting
tf.io.TFRecordCompressionType = tf.io.TFRecordCompressionType.GZIP

Unit tests are still working
Maybe a new test can be added for the gzip case

jhseu · 2019-03-14T20:17:10Z

I'm ok with a pull request that sets the TFRecordCompressionType. I'm just worried about using different compression formats. The TFRecord compression is integrated into the file format, and this PR gzips the file which ends up being a different format. Happy to accept a pull request that can set the option in TFRecords.

fhoering · 2019-03-15T10:04:39Z

@jhseu

Can you give more details on

the TFRecord compression is integrated into the file format

What do you mean by ?

and this PR gzips the file which ends up being a different format

gzip is standardized. So it always will be the gzip format.

From how I understand it (please correct me if I'm wrong)

TFRecord is a protobuf with schema Example or SequenceExample
Protobuf doesn't have integrated gzip compression
If GZIP compression is integrated in the TfRecordWriter Python API it means that there is (somehow) a c++ Gzip outstream class on top of the encoded protobuf
Here we are in the Java world, TFRecordWriter is writing the Java protobuf Example/SequenceExample
We need to somehow gzip the protobuf output stream before writing to disk
So it ends up applying a Java GzipOutputStream on top of that
This is exactly what is proposed here

About the settings, it ends up setting this to the Spark/PySpark job which also seems fine to me.

spark.hadoop.mapreduce.output.fileoutputformat.compress:true
spark.hadoop.mapreduce.output.fileoutputformat.compress.codec:
org.apache.hadoop.io.compress.GzipCodec

Indeed it might not work in some cases as Python api might not support it but it gives enough flexibility for other codecs.
Adding yet another config like tfrecordcompressiontype=Gzip will also be confusing.

At the end on my side I would like to have something that works (and I don't want to maintain my own fork). So any implementation is fine for me.
As indicated I would also provide a new PR. But what is proposed here (fixing using + units tests) looks fine to me.

jhseu · 2019-03-15T17:11:43Z

TFRecord is a format that is essentially a sequence of records stored as bytes. Those can be anything, but are typically tf.Example protos.

The TFRecord has custom compression settings that are different than just gzipping the file, which you can set in the options:
https://www.tensorflow.org/versions/r2.0/api_docs/python/tf/io/TFRecordOptions?hl=en

I'm worried about offering an incompatible, but similar compression setting here. If we could make it compatible, then that's fine.

jhseu · 2019-03-15T17:13:15Z

See the code here:
https://github.com/tensorflow/tensorflow/blob/master/tensorflow/core/lib/io/record_writer.cc

fhoering · 2019-03-15T19:38:42Z

In my opinion what is important is that TfRecords that are written here can be read by Pythons TFRecordReader.
And this is the case for GZIP compression. I have tested it.
I could be a first commit. This will already be fine for 90% of the users.

And when someone wants to implement all options of ZLIB he can do this in a second step. I would argue user rather want to use a newer compression lib than doing that.

jhseu · 2019-03-15T20:03:03Z

It would also need to work with tf.data.

fhoering · 2019-03-15T20:03:51Z

So maybe adding a new setting 'tfrecord.compressiontype=Gzip' is better because it restrains the number of possibilities and can be extended to what has been implemented in TfRecordOptions

fhoering · 2019-03-25T10:32:13Z

I tried this:

tf.data.TFRecordDataset(files, compression_type="GZIP")

It is working.

As suggested I can provide a new PR with the same code + a unit test.
If this is not enough can you please clearly give a timeline on when you will be implementing this on your side or at least give some clear guidelines on what needs to be done or clearly say it will never be done (and close the ticket #61)

jhseu · 2019-03-25T17:56:48Z

@fhoering
Hmm, yeah, it seems I misunderstood the format, and that does seem like it should work after reading through the code more closely. Feel free to send a PR, preferably with a unit test if you can get one to work (if not, it's fine).

fhoering · 2019-03-26T18:17:22Z

@jhseu
I create a new PR https://github.com/tensorflow/ecosystem/pull/125/commits
Please review.

bigbear2017 · 2019-03-27T08:31:20Z

hmmm.... I thought I would have a merge in this project. T.T

bigbear2017 · 2019-03-27T08:32:24Z

anyway, it is good that this has been fixed. : )

jhseu · 2019-03-27T17:19:34Z

Thanks @bigbear2017 for starting this effort! The test that checked that tf.data reads it correctly alleviated my concern about merging this kind of change in.

qiaoguan · 2019-09-06T03:11:05Z

Execuse me, i user spark to transform the original format to TFrecord format, it works , but the tfrecord file is too large , i want to compress it, what should i do?

import org.tensorflow.hadoop.io.TFRecordFileOutputFormat
import org.tensorflow.hadoop.shaded.protobuf.ByteString
import org.apache.hadoop.io.compress.CompressionCodec
import org.apache.hadoop.io.compress.GzipCodec

val tfrecord_sample_dir = sample_dir + localArgs.yyyyMMdd + "/" + localArgs.loc+"_tfrecord"
FileSystem.get(sc.hadoopConfiguration).delete(new Path(tfrecord_sample_dir), true)
tfrecord_sample.saveAsNewAPIHadoopFileTFRecordFileOutputFormat

where tfrecord_sample is a RDD[(BytesWritable, NullWriteable)] format, BytesWritable is a tfrecord example format。 i see saveAsHadoopFile function provide a way to compress the original data, but i do not know how to use it, anyone can help?
/**

Output the RDD to any Hadoop-supported file system, using a Hadoop OutputFormat class
supporting the key and value types K and V in this RDD. Compress the result with the
supplied codec.
*/
def saveAsHadoopFile[F <: OutputFormat[K, V]](
path: String,
codec: Class[_ <: CompressionCodec])(implicit fm: ClassTag[F]): Unit = self.withScope {
val runtimeClass = fm.runtimeClass
saveAsHadoopFile(path, keyClass, valueClass, runtimeClass.asInstanceOf[Class[F]], codec)
}
i try to use
tfrecord_sample.saveAsHadoopFile[TFRecordFileOutputFormat](tfrecord_sample_dir, GzipCodec)
like this , but it doesn't work

nnnnwang · 2020-01-10T14:18:30Z

hello,when i use follow code to compress the tfrecord as gzip format,it seems doesn't work. There is no different between the output file size whether add the "option("codec","org.apache.hadoop.io.compress.GzipCodec")" or not

write.mode("overwrite")
.format("tfrecords")
.option("recordType","Example")
.option("codec", "org.apache.hadoop.io.compress.GzipCodec")
.save(outputFilePath)

fix tfrecord compress on hdfs

8402e2e

googlebot added the cla: no label Oct 30, 2018

googlebot added cla: yes and removed cla: no labels Oct 30, 2018

bigbear2017 mentioned this pull request Oct 30, 2018

[TFRecords File is too large] 3 times than parquet, 1.3 times than normal textFile #106

Closed

bigbear2017 mentioned this pull request Nov 30, 2018

GZIP/ZLIB compression for TFRecord files. #61

Open

bigbear2017 closed this Jan 6, 2019

fhoering reviewed Mar 14, 2019

View reviewed changes

jhseu reopened this Mar 25, 2019

jhseu closed this Mar 27, 2019

		@@ -32,20 +32,29 @@ public class TFRecordFileOutputFormat extends FileOutputFormat<BytesWritable, Nu
		@Override public RecordWriter<BytesWritable, NullWritable> getRecordWriter(

Conversation

bigbear2017 commented Oct 30, 2018

Uh oh!

googlebot commented Oct 30, 2018

What to do if you already signed the CLA

Individual signers

Corporate signers

Uh oh!

bigbear2017 commented Oct 30, 2018

Uh oh!

googlebot commented Oct 30, 2018

Uh oh!

jhseu commented Nov 13, 2018

Uh oh!

bigbear2017 commented Nov 30, 2018

Uh oh!

sahilbadyal commented Mar 8, 2019

Uh oh!

bigbear2017 commented Mar 10, 2019

Uh oh!

sahilbadyal commented Mar 13, 2019

Uh oh!

fhoering commented Mar 14, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

fhoering Mar 14, 2019

Choose a reason for hiding this comment

Uh oh!

fhoering Mar 14, 2019

Choose a reason for hiding this comment

Uh oh!

fhoering commented Mar 14, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jhseu commented Mar 14, 2019

Uh oh!

fhoering commented Mar 15, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jhseu commented Mar 15, 2019

Uh oh!

jhseu commented Mar 15, 2019

Uh oh!

fhoering commented Mar 15, 2019

Uh oh!

jhseu commented Mar 15, 2019

Uh oh!

fhoering commented Mar 15, 2019

Uh oh!

fhoering commented Mar 25, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jhseu commented Mar 25, 2019

Uh oh!

fhoering commented Mar 26, 2019

Uh oh!

bigbear2017 commented Mar 27, 2019

Uh oh!

bigbear2017 commented Mar 27, 2019

Uh oh!

jhseu commented Mar 27, 2019

Uh oh!

qiaoguan commented Sep 6, 2019

Uh oh!

nnnnwang commented Jan 10, 2020

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

7 participants

fhoering commented Mar 14, 2019 •

edited

Loading

fhoering commented Mar 14, 2019 •

edited

Loading

fhoering commented Mar 15, 2019 •

edited

Loading

fhoering commented Mar 25, 2019 •

edited

Loading