在apache_beam中反序列化tfrecords的方法是什么

Question

我确实在apache_beam中找到了专门用于tfrecords的io函数，它可以：

from apache_beam.io.tfrecordio import ReadFromTFRecord
class VerifyOutput(beam.DoFn):
    def process(self, pcollection):
        try:
            pcollection = pcollection.element
        except AttributeError:
            pass
        logging.info(pcollection.subject_id)


(pipeline |  ReadFromTFRecord(opt.input_path, compression_type='auto', validate=True) 
          | beam.ParDo(VerifyOutput()) )

这将打印出tfrecord的bytestringa。如何在beam管道中解析它？这样就可以拿回单个元素了？在

我确实找到了用于读取tfrecordshere的tensorflow习惯用法，但这对我不起作用，我想是因为beam不是“线程安全的”（永远挂起）。在

另外，是否可以（对于调试/学习/测试）在不实际运行管道的情况下运行ReadFromTFRecord？例如在一个jupyter笔记本里？在

Answer 1

您需要指定编码器以解码回特性。 像这样的东西应该可以做到：

import tensorflow as tf
import tensorflow_transform as tft
from apache_beam.io.tfrecordio import ReadFromTFRecord
from tensorflow_transform.beam import impl as beam_impl
from tensorflow_transform.beam import tft_beam_io 
from tensorflow_transform.tf_metadata import dataset_metadata
from tensorflow_transform.tf_metadata import dataset_schema

...

# for each feature, adapt accordingly:
column_schemas['FEATURE_NAME']= dataset_schema.ColumnSchema(tf.int64, [], dataset_schema.FixedColumnRepresentation())

raw_data_metadata = dataset_metadata.DatasetMetadata(dataset_schema.Schema(column_schemas))
data_coder = tft.coders.ExampleProtoCoder(raw_data_metadata.schema)

_ = (pipeline | ReadFromTFRecord(opt.input_path, coder=data_coder, compression_type='auto', validate=True) 
              | beam.ParDo(VerifyOutput()) )

另请参见here以获取更详细的示例。在

但是，考虑到还有（更新的）替代方法来处理TF记录文件和数据集（例如，参见here和here）。在