В предыдущих сообщениях я представил несколько простых развертываний с использованием Tensorflow. В этом посте больше внимания будет уделено преобразованию Tensorflow и Apache Beam. В репозитории TensorFlow Transform Github я опубликовал некоторые проблемы (и решения), связанные с census_example. Поскольку вы можете обратиться к этой ссылке для получения более подробной информации, я не буду публиковать весь код, только то, что необходимо для пояснительных целей.
В этом примере используются данные переписи, чтобы предсказать, сможет ли конкретный человек заработать больше или равно 50 тысяч или меньше 50 тысяч. Входные функции подразделяются на:
CATEGORICAL_FEATURE_KEYS,
NUMERIC_FEATURE_KEYS,
OPTIONAL_NUMERIC_FEATURE_KEYS.
Нам нужно это знать, потому что нам нужно преобразовать данные в соответствующие типы данных с помощью преобразования Tensorflow.
RAW_DATA_FEATURE_SPEC = dict([(name, tf.io.FixedLenFeature([], tf.string)) RAW_DATA_METADATA = dataset_metadata.DatasetMetadata(for name schema_utils.schema_from_feature_spec(RAW_DATA_FEATURE_SPEC)) in CATEGORICAL_FEATURE_KEYS] + [(name, tf.io.FixedLenFeature([], tf.float32)) for name in NUMERIC_FEATURE_KEYS] + [(name, tf.io.VarLenFeature(tf.float32)) for name in OPTIONAL_NUMERIC_FEATURE_KEYS] + [(LABEL_KEY, tf.io.FixedLenFeature([], tf.string))])
Нам нужно это знать, потому что нам нужно преобразовать данные в соответствующие типы данных с помощью преобразования Tensorflow.
RAW_DATA_FEATURE_SPEC.
На основе спецификаций функций функция tf_metadata tensorflow_transform создает схему входных типов объектов. Характеристики и схема функций выглядят следующим образом:
def preprocessing_fn(inputs): outputs = inputs.copy() for key in NUMERIC_FEATURE_KEYS: outputs[key] = tft.scale_to_0_1(outputs[key]) for key in OPTIONAL_NUMERIC_FEATURE_KEYS: dense = tf.compat.v1.sparse_to_dense( outputs[key].indices, [outputs[key].dense_shape[0], 1], outputs[key].values, default_value=0.) dense = tf.squeeze(dense, axis=1) outputs[key] = tft.scale_to_0_1(dense) for key in CATEGORICAL_FEATURE_KEYS: tft.vocabulary(inputs[key], vocab_filename=key) table_keys = ['>50K', '<=50K'] initializer = tf.lookup.KeyValueTensorInitializer( keys=table_keys, values=tf.cast(tf.range(len(table_keys)), tf.int64), key_dtype=tf.string, value_dtype=tf.int64) table = tf.lookup.StaticHashTable(initializer, default_value=-1) outputs[LABEL_KEY] = table.lookup(outputs[LABEL_KEY]) return outputs
Самый важный аспект, который следует учитывать перед обучением модели с этими данными, - это преобразование данных. Иногда это называют функцией предварительной обработки. Мы передадим это во время обучения как функцию-оболочку.
После этого мы будем использовать Apache Beam для преобразования наших обучающих или оценочных данных в TFRecord. Apache Beam - это высокоуровневый API, созданный для обработки пакетных данных и потоковой передачи данных для быстрой обработки (здесь не будем вдаваться в подробности, поскольку это для другой публикации):
with beam.Pipeline() as pipeline:
with tft_beam.Context(temp_dir=tempfile.mktemp()):
ordered_columns = ['age', 'workclass', 'fnlwgt','education', 'education-num','marital-status', 'occupation', 'relationship', 'race', 'sex','capital-gain', 'capital-loss', 'hours-per-week', 'native-country','label']
converter = tft.coders.CsvCoder(ordered_columns, RAW_DATA_METADATA.schema)
raw_data = (
pipeline
| "ReadTrainData" >> beam.io.ReadFromText(train_data_file)
| "FixCommasTrainData" >> beam.Map(lambda line: line.replace(', ', ','))
| "DecodeTrainData" >> MapAndFilterErrors(converter.decode)
)
raw_dataset = (raw_data, RAW_DATA_METADATA)
transformed_dataset , transform_fn = (raw_dataset | tft_beam.AnalyzeAndTransformDataset(preprocessing_fn))
transformed_data, transformed_metadata = transformed_dataset
transformed_data_coder = tft.coders.ExampleProtoCoder(transformed_metadata.schema)
_ = (
transformed_data
| "EncodeTrainData" >> beam.Map(transformed_data_coder.encode)
| "WriteTrainData" >> beam.io.WriteToTFRecord(os.path.join(working_dir, TRANSFORMED_TRAIN_DATA_FILEBASE))
)
Функция ReadFromText Apache Beam будет читать текстовый файл, исправлять запятые в каждой строке (записи) и использовать класс MapAndFilterErrors для декодирования данных в формате .csv. Затем мы можем очень просто создать набор данных с необработанными данными и схемой, как показано выше. Затем мы хотим выполнить преобразование с помощью AnalyzeAndTransformDataset с нашей функцией предварительной обработки в качестве оболочки. Наконец, мы записываем преобразованные данные на диск в виде файла типа данных TFRecord. Мы делаем то же самое для набора оценочных данных.
Здесь мы покажем, как создать обучающую оболочку с преобразованным набором данных в качестве входного параметра:
def _make_training_input_fn(tf_transform_output, transformed_examples, batch_size): def input_fn(): dataset = tf.data.experimental.make_batched_features_dataset( file_pattern=transformed_examples, batch_size=batch_size, features=tf_transform_output.transformed_feature_spec(), reader=tf.data.TFRecordDataset, shuffle=) transformed_features = tf.compat.v1.data.make_one_shot_iterator(dataset).get_next() transformed_labels = transformed_features.pop(LABEL_KEY) return transformed_features, transformed_labels return input_fn
Мы также можем создать оболочку для функции обслуживания, которую можно использовать как во время оценки, так и во время последующего обслуживания. Разница в том, что нам не нужны функции ярлыков, которые противоречат цели обучения!
Здесь мы используем API оценщика для обучения, и мы используем LinearClassifier, но не стесняйтесь использовать любой другой, например DNNClassifier:
estimator = tf.estimator.LinearClassifier( feature_columns=get_feature_columns(tf_transformed_output), config=run_config, loss_reduction=tf.losses.Reduction.SUM ) train_input_fn = _make_training_input_fn( tf_transformed_output, os.path.join(working_dir, TRANSFORMED_TRAIN_DATA_FILEBASE + '*'), batch_size=TRAIN_BATCH_SIZE ) estimator.train(input_fn=train_input_fn, max_steps=TRAIN_NUM_EPOCHS * num_train_instances / TRAIN_BATCH_SIZE)
После завершения обучения у нас будет модель вывода, сохраненная в нашем working_dir, и мы сможем проверить определение сигнатуры нашей сохраненной модели, используя saved_model_cli TensorFlow.
рисунок: определение сигнатуры сохраненной модели
Теперь мы готовы обслуживать модель, обученную на основе набора данных переписи, с использованием обслуживания TensorFlow.
рисунок: обслуживание модели с использованием модели Tensorflow
Конечно, неплохо создать клиента для вывода. В предыдущих сообщениях я использовал python для создания клиента для вывода. На самом деле мы можем использовать Curl для вывода, и это самый простой способ сделать это:
$ curl -d '{"examples": [{"age":30.0, "workclass":"Self-emp-not-inc", "education":"Bachelors", "education-num":17.0, "marital-status":"Married-civ-spouse", "occupation":"Exec-managerial", "relationship":"Husband", "race":"White", "sex":"Male", "capital-gain":0.0, "capital-loss":0.0, "hours-per-week":40.0, "native-country":"United-States"}]}' -X POSThttp://localhost:8501/v1/models/census:classify
{
"results": [[["0", 0.498906225], ["1", 0.501093805]]
]
Надеюсь, вам понравились эти сообщения.
Tensorflow и Tensorflow Extended - мощные инструменты. В этой серии постов я попытался показать несколько простых реализаций, которые помогут вам начать использовать эти инструменты.
Пожалуйста, присылайте нам свои комментарии, исправления и дальнейшие обсуждения по адресу [email protected]
Готовы узнать больше?
Свяжитесь с нашей командой экспертов, чтобы узнать, как мы можем помочь вам в создании интеллектуальных решений для прогнозирования вашего бизнеса.
