neuromage
commented
5 years ago For reference, here's a simple pipeline you can run (it's using TFX DSL) which will output some basic metadata in your cluster:
import argparse
import os
import tensorflow as tf
from typing import Text
import kfp
from tfx.components.evaluator.component import Evaluator
from tfx.components.example_gen.csv_example_gen.component import CsvExampleGen
from tfx.components.example_validator.component import ExampleValidator
from tfx.components.model_validator.component import ModelValidator
from tfx.components.pusher.component import Pusher
from tfx.components.schema_gen.component import SchemaGen
from tfx.components.statistics_gen.component import StatisticsGen
from tfx.components.trainer.component import Trainer
from tfx.components.transform.component import Transform
from tfx.orchestration import metadata
from tfx.orchestration import pipeline
from tfx.orchestration.kubeflow import kubeflow_dag_runner
from tfx.proto import evaluator_pb2
from tfx.utils.dsl_utils import csv_input
from tfx.proto import pusher_pb2
from tfx.proto import trainer_pb2
from tfx.extensions.google_cloud_ai_platform.trainer import executor as ai_platform_trainer_executor
_output_bucket = 'gs://your-bucket-here'
def _create_test_pipeline(pipeline_name: Text, pipeline_root: Text,
csv_input_location: Text, taxi_module_file: Text)
"""Creates a simple Kubeflow-based Chicago Taxi TFX pipeline for testing.
Args:
pipeline_name: The name of the pipeline.
pipeline_root: The root of the pipeline output.
csv_input_location: The location of the input data directory.
taxi_module_file: The location of the module file for Transform/Trainer.
container_image: The container image to use.
Returns:
A logical TFX pipeline.Pipeline object.
"""
examples = csv_input(csv_input_location)
example_gen = CsvExampleGen(input_base=examples)
statistics_gen = StatisticsGen(input_data=example_gen.outputs.examples)
infer_schema = SchemaGen(
stats=statistics_gen.outputs.output, infer_feature_shape=False)
validate_stats = ExampleValidator(
stats=statistics_gen.outputs.output, schema=infer_schema.outputs.output)
transform = Transform(
input_data=example_gen.outputs.examples,
schema=infer_schema.outputs.output,
module_file=taxi_module_file)
trainer = Trainer(
module_file=taxi_module_file,
transformed_examples=transform.outputs.transformed_examples,
schema=infer_schema.outputs.output,
transform_output=transform.outputs.transform_output,
train_args=trainer_pb2.TrainArgs(num_steps=10000),
eval_args=trainer_pb2.EvalArgs(num_steps=5000))
model_analyzer = Evaluator(
examples=example_gen.outputs.examples,
model_exports=trainer.outputs.output,
feature_slicing_spec=evaluator_pb2.FeatureSlicingSpec(specs=[
evaluator_pb2.SingleSlicingSpec(
column_for_slicing=['trip_start_hour'])
]))
model_validator = ModelValidator(
examples=example_gen.outputs.examples, model=trainer.outputs.output)
pusher = Pusher(
model_export=trainer.outputs.output,
model_blessing=model_validator.outputs.blessing,
push_destination=pusher_pb2.PushDestination(
filesystem=pusher_pb2.PushDestination.Filesystem(
base_directory=os.path.join(pipeline_root, 'model_serving'))))
return pipeline.Pipeline(
pipeline_name=pipeline_name,
pipeline_root=pipeline_root,
components=[
example_gen, statistics_gen, infer_schema, validate_stats, transform,
trainer, model_analyzer, model_validator, pusher
],
enable_cache=False, # Or True to use cache
)
if __name__ == '__main__':
# Copy sample CSV file from chicago taxi pipeline example to this location
data_root = 'gs://your-bucket/data'
taxi_module_file = 'gs://your-bucket/taxi_utils.py'
pipeline_name = 'kubeflow-simple-taxi-metadata'
pipeline_root = 'gs://your-bucket/test'
pipeline = _create_test_pipeline(pipeline_name, pipeline_root, data_root, taxi_module_file)
config = kubeflow_dag_runner.KubeflowRunnerConfig()
kubeflow_dag_runner.KubeflowDagRunner(config=config).run(pipeline)
Bobgy
k8s-ci-robot
rmgogogo
dushyanthsc
Now that metadata pages are in KFP's UI thanks to Riley's work, there are still a few more items to take care of in terms of polish:
We should add a column showing the creation time of each artifact. In order to get this, we'll need to query the metadata API server for events, and find the associated event that produced any given artifact. The event will have a timestamp.
The list of executions page seems a little buggy. When I click on an execution, the first one in the group (they are grouped by pipeline) works, but the following items don't seem to be working.
For each execution, other than properties, we should also show the inputs and outputs that went into it. It would also be nice to be able to link to the said input and output.
Assigning to Yuan to start work on this. I'll update this issue with any other outstanding items I find. Yuan, we can also chat in person to clarify these items as required. Thanks!
/assign @Bobgy
/cc @jessiezcc /cc @paveldournov /cc @dushyanthsc /cc @gaoning777