<jupyter_start><jupyter_text>We start by downloading a specific release of the components because running from master is not a good way to buid "repetable" systems<jupyter_code>!wget https://github.com/kubeflow/pipelines/archive/0.2.5.tar.gz
!tar -xvf 0.2.5.tar.gz
import kfp
#tag::loadGCSDLComponent[]
gcs_download_component = kfp.components.load_component_from_file(
    "pipelines-0.2.5/components/google-cloud/storage/download/component.yaml")
#end::loadGCSDLComponent[]
#tag::loadTFDVAndFriendsComponents[]
tfx_csv_gen = kfp.components.load_component_from_file(
    "pipelines-0.2.5/components/tfx/ExampleGen/CsvExampleGen/component.yaml")
tfx_statistic_gen = kfp.components.load_component_from_file(
    "pipelines-0.2.5/components/tfx/StatisticsGen/component.yaml")
tfx_schema_gen = kfp.components.load_component_from_file(
    "pipelines-0.2.5/components/tfx/SchemaGen/component.yaml")
tfx_example_validator = kfp.components.load_component_from_file(
    "pipelines-0.2.5/components/tfx/ExampleValidator/component.yaml")
#end::loadTFDVAndFriendsComponents[]
@kfp.dsl.pipeline(
  name='DL',
  description='Sample DL pipeline'
)
def pipeline_with_dl():
    #tag::dlOp[]
    dl_op = gcs_download_component(
        gcs_path="gs://ml-pipeline-playground/tensorflow-tfx-repo/tfx/components/testdata/external/csv") # Your path goes here
    #end::dlOp[]
kfp.compiler.Compiler().compile(pipeline_with_dl, 'dl_pipeline.zip')
client = kfp.Client()
my_experiment = client.create_experiment(name='dl')
my_run = client.run_pipeline(my_experiment.id, 'dl', 
  'dl_pipeline.zip')
#tag::standaloneTFDVPipeline[]
@kfp.dsl.pipeline(
  name='TFDV',
  description='TF DV Pipeline'
)
def tfdv_pipeline():
    # DL with wget, can use gcs instead as well
    data_url = "https://raw.githubusercontent.com/moorissa/medium/master/items-recommender/data/trx_data.csv"
    #tag::wget[]
    fetch = kfp.dsl.ContainerOp(
      name='download',
      image='busybox',
      command=['sh', '-c'],
      arguments=[
          'sleep 1;'
          'mkdir -p /tmp/data;'
          'wget '+ data_url +' -O /tmp/data/results.csv'],
      file_outputs={'downloaded': '/tmp/data'})
    # This expects a directory of inputs not just a single file
    #end::wget[]
    #tag::csv[]
    records_example = tfx_csv_gen(input_base=fetch.output)
    #end::csv[]
    #tag::stats[]
    stats = tfx_statistic_gen(input_data=records_example.output)
    #end::stats[]
    #tag::schema[]
    schema_op = tfx_schema_gen(stats.output)
    #end::schema[]
    #tag::validate[]
    tfx_example_validator(stats=stats.outputs['output'], schema=schema_op.outputs['output'])
    #end::validate[]
#end::standaloneTFDVPipeline[]
kfp.compiler.Compiler().compile(tfdv_pipeline, 'tfdv_pipeline.zip')
my_experiment = client.create_experiment(name='tfdv_pipeline')
my_run = client.run_pipeline(my_experiment.id, 'tfdv', 
  'tfdv_pipeline.zip')
!pip3 install tfx tensorflow-data-validation
#tag::importTFDV[]
import tensorflow_data_validation as tfdv
#end::importTFDV[]<jupyter_output><empty_output><jupyter_text>You can download your schema by looking at the inputs/outputs in your pipeline run for the schema gen stage<jupyter_code>#tag::displaySchema{}
schema = tfdv.load_schema_text("schema_info_2")
tfdv.display_schema(schema)
#end::displaySchema[]
#tag::loadTFT[]
tfx_transform = kfp.components.load_component_from_file(
    "pipelines-0.2.5/components/tfx/Transform/component.yaml")
#end::loadTFT[]
module_file="gcs://"
@kfp.dsl.pipeline(
  name='TFX',
  description='TFX pipeline'
)
def tfx_pipeline():
    # DL with wget, can use gcs instead as well
    fetch = kfp.dsl.ContainerOp(
      name='download',
      image='busybox',
      command=['sh', '-c'],
      arguments=[
          'sleep 1;'
          'mkdir -p /tmp/data;'
          'wget https://raw.githubusercontent.com/moorissa/medium/master/items-recommender/data/trx_data.csv -O /tmp/data/results.csv'],
      file_outputs={'downloaded': '/tmp/data'})
    records_example = tfx_csv_gen(input_base=fetch.output)
    stats = tfx_statistic_gen(input_data=records_example.output)
    schema_op = tfx_schema_gen(stats.output)
    tfx_example_validator(stats=stats.outputs['output'], schema=schema_op.outputs['output'])
    #tag::tft[]
    transformed_output = tfx_transform(
        input_data=records_example.output,
        schema=schema_op.outputs['output'],
        module_file=module_file) # Path to your TFT code on GCS/S3
    #end::tft[]
kfp.compiler.Compiler().compile(tfx_pipeline, 'tfx_pipeline.zip')
my_experiment = client.create_experiment(name='tfx_pipeline')
my_run = client.run_pipeline(my_experiment.id, 'tfx', 
  'tfx_pipeline.zip')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14047822374039282, 'fraction_numerical': 0.006191289496157131, 'mean_word_length': 4.900503778337532, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 9, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '438541', 'license': 'Apache-2.0', 'n_tokens_mistral': 1664, 'n_tokens_neox': 1665, 'n_words': 250, 'path': 'data-extraction/tfx/TFDV.ipynb', 'repo_name': 'gabrielclimb/intro-to-ml-with-kubeflow-examples', 'size': '4684'}