python - 应用 TensorFlow Transform 来转换/缩放生产中的特征

Question

概览

我按照以下指南编写了 TF Records，其中我使用了 tf.Transform 来预处理我的功能。现在，我想部署我的模型，为此我需要对实时数据应用此预处理功能。

我的方法

首先，假设我有两个特征:

features = ['amount', 'age']

我有来自 Apache Beam 的 transform_fn，位于 working_dir=gs://path-to-transform-fn/

然后我使用以下方法加载转换函数:

tf_transform_output = tft.TFTransformOutput(working_dir)

我认为在生产中提供服务的最简单方法是获取经过处理的数据的 numpy 数组，然后调用 model.predict()(我使用的是 Keras 模型)。

为此，我想 transform_raw_features()方法正是我需要的。

但是，似乎在构建模式之后:

raw_features = {}
for k in features:
    raw_features.update({k: tf.constant(1)})

print(tf_transform_output.transform_raw_features(raw_features))

我得到:

AttributeError: 'Tensor' object has no attribute 'indices'

现在，我假设发生这种情况是因为我在 preprocessing_fn 中定义架构时使用了 tf.VarLenFeature()。

def preprocessing_fn(inputs):
    outputs = inputs.copy()

    for _ in features:
        outputs[_] = tft.scale_to_z_score(outputs[_])

我使用以下方法构建元数据:

RAW_DATA_FEATURE_SPEC = {}
for _ in features:
    RAW_DATA_FEATURE_SPEC[_] = tf.VarLenFeature(dtype=tf.float32)
    RAW_DATA_METADATA = dataset_metadata.DatasetMetadata(
    dataset_schema.from_feature_spec(RAW_DATA_FEATURE_SPEC))

简而言之，给定一个字典:

d = {'amount': [50], 'age': [32]}，我想应用这个 transform_fn，并适本地缩放这些值输入到我的模型中进行预测。这个字典正是我的PCollection在数据被pre_processing()函数处理之前的格式。

管道结构:

class BeamProccess():

def __init__(self):

    # init 

    self.run()


def run(self):

    def preprocessing_fn(inputs):

         # outputs = { 'id' : [list], 'amount': [list], 'age': [list] }
         return outputs

    with beam.Pipeline(options=self.pipe_opt) as p:
        with beam_impl.Context(temp_dir=self.google_cloud_options.temp_location):
            data = p | "read_table" >> beam.io.Read(table_bq) \
            | "create_data" >> beam.ParDo(ProcessFn())

            transformed_dataset, transform_fn = (
                        (train, RAW_DATA_METADATA) | beam_impl.AnalyzeAndTransformDataset(
                    preprocessing_fn))

            transformed_data, transformed_metadata = transformed_dataset

            transformed_data | "WriteTrainTFRecords" >> tfrecordio.WriteToTFRecord(
                    file_path_prefix=self.JOB_DIR + '/train/data',
                    file_name_suffix='.tfrecord',
                    coder=example_proto_coder.ExampleProtoCoder(transformed_metadata.schema))

            _ = (
                        transform_fn
                        | 'WriteTransformFn' >>
                        transform_fn_io.WriteTransformFn(path=self.JOB_DIR + '/transform/'))

最后 ParDo() 是:

class ProcessFn(beam.DoFn):

    def process(self, element):

        yield { 'id' : [list], 'amount': [list], 'age': [list] }

Answer 1

问题出在片段上

raw_features = {}
for k in features:
    raw_features.update({k: tf.constant(1)})

print(tf_transform_output.transform_raw_features(raw_features))

在这段代码中，您构建了一个字典，其中的值是张量。正如您所说，这不适用于 VarLenFeature。而不是使用 tf.constant 尝试使用 tf.placeholder 作为 FixedLenFeature 和 tf.sparse_placeholder 作为 VarLenFeature.