5

我对 TFX 很陌生，但有一个显然可以通过BulkInferrer使用的 ML 管道。这似乎只以 Protobuf 格式产生输出，但由于我正在运行批量推理，我想将结果通过管道传输到数据库。（DB 输出似乎应该是批量推理的默认值，因为批量推理和 DB 访问都利用了并行化......但 Protobuf 是每记录的序列化格式。）

我假设我可以使用Parquet-Avro-Protobuf之类的东西来进行转换（尽管这是在 Java 中，而管道的其余部分在 Python 中），或者我可以自己编写一些东西来一个接一个地使用所有 protobuf 消息，转换将它们转换为 JSON，将 JSON 反序列化为字典列表，然后将字典加载到 Pandas DataFrame 中，或者将其存储为一堆键值对，我将其视为一次性数据库......但这听起来像对于一个非常常见的用例，涉及并行化和优化的大量工作和痛苦。顶级 Protobuf 消息定义是 Tensorflow 的PredictionLog。

这一定是一个常见的用例，因为像这样的 TensorFlowModelAnalytics 函数使用 Pandas DataFrames。我宁愿能够直接写入数据库（最好是 Google BigQuery）或 Parquet 文件（因为 Parquet / Spark 似乎比 Pandas 并行化更好），而且这些似乎应该是常见的用例，但我没有找到任何例子。也许我使用了错误的搜索词？

我还查看了PredictExtractor，因为“提取预测”听起来很接近我想要的......但官方文档似乎没有说明应该如何使用该类。我认为TFTransformOutput听起来像是一个很有前途的动词，但实际上它是一个名词。

我显然在这里遗漏了一些基本的东西。没有人愿意将 BulkInferrer 结果存储在数据库中吗？是否有允许我将结果写入数据库的配置选项？也许我想将ParquetIO或BigQueryIO实例添加到 TFX 管道？（TFX 文档说它在“幕后”使用 Beam，但这并没有说明我应该如何一起使用它们。）但是这些文档中的语法看起来与我的 TFX 代码完全不同，我不确定它们是否'重新兼容？

帮助？

4

3 回答 3

2

（从相关问题复制以提高知名度）

经过一番挖掘，这是一种替代方法，它假设feature_spec事先不知道。请执行下列操作：

  • 将 设置BulkInferrer为写入，output_examples而不是inference_result将output_example_spec添加到组件构造中。
  • 在主管道中添加一个 StatisticsGen和一个SchemaGen组件，BulkInferrer以生成上述模式的架构output_examples
  • 使用来自SchemaGen和BulkInferrer读取 TFRecords 的工件并做任何必要的事情。
bulk_inferrer = BulkInferrer(
     ....
     output_example_spec=bulk_inferrer_pb2.OutputExampleSpec(
         output_columns_spec=[bulk_inferrer_pb2.OutputColumnsSpec(
             predict_output=bulk_inferrer_pb2.PredictOutput(
                 output_columns=[bulk_inferrer_pb2.PredictOutputCol(
                     output_key='original_label_name',
                     output_column='output_label_column_name', )]))]
     ))

 statistics = StatisticsGen(
     examples=bulk_inferrer.outputs.output_examples
 )

 schema = SchemaGen(
     statistics=statistics.outputs.output,
 )

之后，可以执行以下操作：

import tensorflow as tf
from tfx.utils import io_utils
from tensorflow_transform.tf_metadata import schema_utils

# read schema from SchemaGen
schema_path = '/path/to/schemagen/schema.pbtxt'
schema_proto = io_utils.SchemaReader().read(schema_path)
spec = schema_utils.schema_as_feature_spec(schema_proto).feature_spec

# read inferred results
data_files = ['/path/to/bulkinferrer/output_examples/examples/examples-00000-of-00001.gz']
dataset = tf.data.TFRecordDataset(data_files, compression_type='GZIP')

# parse dataset with spec
def parse(raw_record):
    return tf.io.parse_example(raw_record, spec)

dataset = dataset.map(parse)

此时，数据集就像任何其他已解析的数据集一样，因此编写 CSV 或 BigQuery 表或从那里写入任何内容都是微不足道的。它确实通过BatchInferencePipeline在ZenML中帮助了我们。

于 2021-01-31T12:24:54.603 回答
0

我参加这个聚会有点晚了，但这是我用于此任务的一些代码：

import tensorflow as tf
from tensorflow_serving.apis import prediction_log_pb2
import pandas as pd


def parse_prediction_logs(inference_filenames: List[Text]): -> pd.DataFrame
    """
    Args:
        inference files:  tf.io.gfile.glob(Inferrer artifact uri)
    Returns:
        a dataframe of userids, predictions, and features
    """

    def parse_log(pbuf):
        # parse the protobuf
        message = prediction_log_pb2.PredictionLog()
        message.ParseFromString(pbuf)
        # my model produces scores and classes and I extract the topK classes
        predictions = [x.decode() for x in (message
                                            .predict_log
                                            .response
                                            .outputs['output_2']
                                            .string_val
                                            )[:10]]
        # here I parse the input tf.train.Example proto
        inputs = tf.train.Example()
        inputs.ParseFromString(message
                               .predict_log
                               .request
                               .inputs['input_1'].string_val[0]
                               )

        # you can pull out individual features like this         
        uid = inputs.features.feature["userId"].bytes_list.value[0].decode()

        feature1 = [
            x.decode() for x in inputs.features.feature["feature1"].bytes_list.value
        ]

        feature2 = [
            x.decode() for x in inputs.features.feature["feature2"].bytes_list.value
        ]

        return (uid, predictions, feature1, feature2)

    return pd.DataFrame(
        [parse_log(x) for x in
         tf.data.TFRecordDataset(inference_filenames, compression_type="GZIP").as_numpy_iterator()
        ], columns = ["userId", "predictions", "feature1", "feature2"]
    )
于 2022-02-04T22:16:53.440 回答
0

在这里回答我自己的问题以记录我们所做的事情，尽管我认为下面@Hamza Tahir 的回答客观上更好。这可以为需要更改开箱即用 TFX 组件的操作的其他情况提供一个选项。虽然它很hacky：

_run_model_inference()我们复制并编辑了文件 tfx/components/bulk_inferrer/executor.py，在方法的内部管道中替换了这个转换：

| 'WritePredictionLogs' >> beam.io.WriteToTFRecord(
             os.path.join(inference_result.uri, _PREDICTION_LOGS_FILE_NAME),
             file_name_suffix='.gz',
             coder=beam.coders.ProtoCoder(prediction_log_pb2.PredictionLog)))

有了这个：

| 'WritePredictionLogsBigquery' >> beam.io.WriteToBigQuery(
           'our_project:namespace.TableName',
           schema='SCHEMA_AUTODETECT',
           write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
           create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
           custom_gcs_temp_location='gs://our-storage-bucket/tmp',
           temp_file_format='NEWLINE_DELIMITED_JSON',
           ignore_insert_ids=True,
       )

（这是有效的，因为当您导入 BulkInferrer 组件时，每个节点的工作被外包给在工作节点上运行的这些执行程序，并且 TFX 将其自己的库复制到这些节点上。它不会从用户空间库中复制所有内容，不过，这就是为什么我们不能只继承 BulkInferrer 并导入我们的自定义版本。）

我们必须确保 at 的表'our_project:namespace.TableName'具有与模型输出兼容的模式，但不必将该模式转换为 JSON / AVRO。

理论上，我的团队希望使用围绕此构建的 TFX 发出拉取请求，但目前我们正在硬编码几个关键参数，并且没有时间将其变为真正的公共/生产状态。

于 2021-02-05T15:31:02.800 回答