TFX core mission is to allow models to be moved from research to production, creating and managing production pipelines. Many models will be built using large volumes of data, requiring multiple hosts working in parallel to serve both the processing and serving needs of your production pipelines.
Using capabilities inherited from Apache Beam, TFX's data processing framework, we will look at how a TFX pipeline, developed against a small dataset, can be scaled out for your production dataset.
Apache Beam
The origins of Apache Beam can be traced back to FlumeJava, which is the data processing framework used at Google (discussed in the ) , the external version of TFX makes use of the external version of Google Flume, Apache Beam.The Apache Beam portable API layer powers TFX libraries (for example , and .
The runner, used in this blog, is Dataflow which shares a large percentage of its code with Google Flume, with further unification in progress.
Below we can see the graph created by the TFX component ExampleGen when it is run on the Dataflow runner.
TFX Libraries
TFX pipeline components are built upon TFX libraries. For example, TensorFlow Transform, which uses Apache Beam. We will explore this library using two different Apache Beam runners. Initially, the local development runner DirectRunner will be used. This will be followed by some minor code modifications to run the sample with the production Dataflow runner. The DirectRunner is a lightweight runner for development purposes. It runs locally and does not require a distributed processing framework.The example pipeline below is borrowed from the tutorial ( can be used to preprocess data.
For details of the
preprocessing_fn, please refer back to the tutorial. For now, we just need to know that it is transforming the data points passed into the function. NOTE:
Environment used for this blog post:
PYTHON
virtualenv tfx-beam --python=python3
source tfx-beam/bin/activate
pip install tfx PYTHON
def main():
with tft_beam.Context(temp_dir=tempfile.mkdtemp()):
transformed_dataset, transform_fn = (
(raw_data, raw_data_metadata) | tft_beam.AnalyzeAndTransformDataset(
preprocessing_fn))
transformed_data, transformed_metadata = transformed_dataset
print('\nRaw data:\n{}\n'.format(pprint.pformat(raw_data)))
print('Transformed data:\n{}'.format(pprint.pformat(transformed_data)))
if __name__ == '__main__':
main(). For example, in this line:
PYTHON
result = pass_this | 'name this step' >> to_this_callto_this_call is being invoked and passed the object called pass_this, and PYTHON
# Setup our Environment
## The location of Input / Output between various stages ( TFX Components )
## This will also be the location for the Metadata
### Can be used when running the pipeline locally
#LOCAL_PIPELINE_ROOT =
### In production you want the input and output to be stored on non-local location
#GOOGLE_CLOUD_STORAGE_PIPELINE_ROOT=
#GOOGLE_CLOUD_PROJECT =
#GOOGLE_CLOUD_TEMP_LOCATION =
# Will need setup.py to make this work with Dataflow
#
# import setuptools
#
# setuptools.setup(
# name='demo',
# version='0.0',
# install_requires=['tfx==0.21.1'],
# packages=setuptools.find_packages(),)
SETUP_FILE = "./setup.py"
argv=['--project={}'.format(GOOGLE_CLOUD_PROJECT),
'--temp_location={}'.format(GOOGLE_CLOUD_TEMP_LOCATION),
'--setup_file={}'.format(SETUP_FILE),
'--runner=DataflowRunner']
def main():
with beam.Pipeline(argv=argv) as p:
with beam_impl.Context(temp_dir=GOOGLE_CLOUD_TEMP_LOCATION):
input = p | beam.Create(raw_data)
transformed_data, transformed_metadata = (
(input, raw_data_metadata)
| beam_impl.AnalyzeAndTransformDataset(preprocessing_fn))
if __name__ == '__main__':
main()allows us to directly query data in BigQuery.
PYTHON
def createExampleGen(query: Text):
# Output 2 splits: train:eval=3:1.
output = example_gen_pb2.Output(
split_config=example_gen_pb2.SplitConfig(splits=[
example_gen_pb2.SplitConfig.Split(
name='train', hash_buckets=3),
example_gen_pb2.SplitConfig.Split(name='eval', hash_buckets=1)
]))
return BigQueryExampleGen(query=query, output_config=output)The data for this example comes from the public Chicago taxi trips dataset located on BigQuery's public datasets (Google Cloud's Data warehouse).
bigquery-public-data.chicago_taxi_trips.taxi_trips. NOTE: You can find more details about BigQuery public datasets at: and That works fine for one hundred records, but what if the goal was to process all 187,002,0025 rows in the dataset? For this, the pipeline is switched from the DirectRunner to the production Dataflow runner. A few extra environment parameters are also set, for example the Google Cloud project to run the pipeline in.
PYTHON
tfx_pipeline = createTfxPipeline(
pipeline_name="my_first_dataflowRunner_pipeline",
pipeline_root=GOOGLE_CLOUD_STORAGE_PIPELINE_ROOT,
query=query,
beam_pipeline_args={
'beam_pipeline_args':[
'--project={}'.format(GOOGLE_CLOUD_PROJECT)
,
'--temp_location={}'.format(GOOGLE_CLOUD_TEMP_LOCATION),
'--setup_file=./setup.py',
'--runner=DataflowRunner']})
BeamDagRunner().run(tfx_pipeline)BeamDagRunner takes care of submitting ExampleGen and StatisticsGen as separate pipelines, ensuring ExampleGen completed successfully first before starting StatisticsGen. The Dataflow service automatically takes care of spinning up workers, autoscaling, retries in the event of worker failure, centralized logging, and monitoring. Autoscaling is based on various signals including throughput rate, illustrated below; Apache Beam supports custom counters, which allows developers to create metrics from within their pipelines. The TFX team has used this to create useful information counters for the various components. Below we can see some of the counters recorded during the StatisticsGen run. Filtering for the key word "num_*_feature", there were roughly a billion integers and float features values. , join the , watch our to the TensorFlow channel. Vollständiger Original-Bericht
Ausführliche Details, Code-Beispiele & Hersteller-Stellungnahme auf blog.tensorflow.org.
SOCIAL SHARE CARD GENERATOR