Обзор
Я следовал следующему руководству, чтобы написать TF Records, где я использовал tf.Transform
для предварительной обработки своих функций. Теперь я хотел бы развернуть свою модель, для которой мне нужно применить эту функцию предварительной обработки к реальным живым данным.
Мой подход
Сначала предположим, что у меня есть 2 функции:
features = ['amount', 'age']
У меня есть transform_fn
от Луча Апача, проживающего в working_dir=gs://path-to-transform-fn/
Затем я загружаю функцию преобразования, используя:
tf_transform_output = tft.TFTransformOutput(working_dir)
Я подумал, что самым простым способом обслуживания в производственной среде было получить массив обработанных данных и вызвать 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'
Теперь, я предполагаю, что это происходит, потому что я использовал tf.VarLenFeature()
, когда я определил схему в своем preprocessing_fn
.
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] }