<jupyter_start><jupyter_text>We are going to be creating a ZenML training pipeline and showcasing the modularity of ZenML backends in this example. On a high level, here is what a ZenML training pipeline looks like:   Set up some variables<jupyter_code>GCP_BUCKET=os.getenv('GCP_BUCKET')
GCP_PROJECT=os.getenv('GCP_PROJECT')
GCP_REGION=os.getenv('GCP_REGION')
GCP_CLOUD_SQL_INSTANCE_NAME=os.getenv('GCP_CLOUD_SQL_INSTANCE_NAME')
MODEL_NAME=os.getenv('MODEL_NAME')
CORTEX_ENV=os.getenv('CORTEX_ENV')
MYSQL_DB=os.getenv('MYSQL_DB')
MYSQL_USER=os.getenv('MYSQL_USER')
MYSQL_PWD=<PASSWORD>('MYSQL_PWD')
MYSQL_PORT=os.getenv('MYSQL_PORT')
MYSQL_HOST=os.getenv('MYSQL_HOST')
CONNECTION_NAME = f'{GCP_PROJECT}:{GCP_REGION}:{GCP_CLOUD_SQL_INSTANCE_NAME}'
TRAINING_JOB_DIR = os.path.join(GCP_BUCKET, 'gcp_gcaip_training/staging')
repo: Repository = Repository.get_instance()
    
# Define artifact store in the cloud
cloud_artifact_store = ArtifactStore(os.path.join(GCP_BUCKET, 'all_feature_demo'))

# Define metadata store in the cloud
cloud_metadata_store = MySQLMetadataStore(
    host=MYSQL_HOST,
    port=int(MYSQL_PORT),
    database=MYSQL_DB,
    username=MYSQL_USER,
    password=<PASSWORD>,
)<jupyter_output><empty_output><jupyter_text>Create first pipeline<jupyter_code>training_pipeline = TrainingPipeline(name='Experiment 1')<jupyter_output><empty_output><jupyter_text>Add a datasource. This will automatically track and version it.<jupyter_code>try:
    ds = CSVDatasource(name='Pima Indians Diabetes', path='gs://zenml_quickstart/diabetes.csv')
except:
    repo: Repository = Repository.get_instance()
    ds = repo.get_datasource_by_name('Pima Indians Diabetes')
training_pipeline.add_datasource(ds)<jupyter_output><empty_output><jupyter_text>Add a split step to partition data into train and eval<jupyter_code>training_pipeline.add_split(RandomSplit(split_map={'train': 0.7, 'eval': 0.2, 'test':0.1}))<jupyter_output><empty_output><jupyter_text>Add a preprocessing step to transform data to be ML-capable<jupyter_code>training_pipeline.add_preprocesser(
    StandardPreprocesser(
        features=['times_pregnant', 'pgc', 'dbp', 'tst', 'insulin', 'bmi',
                  'pedigree', 'age'],
        labels=['has_diabetes'],
        overwrite={'has_diabetes': {
            'transform': [{'method': 'no_transform', 'parameters': {}}]}}
    ))<jupyter_output><empty_output><jupyter_text>Add a trainer which defines model and training<jupyter_code>training_pipeline.add_trainer(TFFeedForwardTrainer(
    loss='binary_crossentropy',
    last_activation='sigmoid',
    output_units=1,
    metrics=['accuracy'],
    epochs=5))<jupyter_output><empty_output><jupyter_text>Add an evaluator to calculate slicing metrics<jupyter_code>training_pipeline.add_evaluator(
    TFMAEvaluator(slices=[['has_diabetes']],
                  metrics={transformed_label_name('has_diabetes'):
                     ['binary_crossentropy', 'binary_accuracy']}))<jupyter_output><empty_output><jupyter_text>Run and evaluate<jupyter_code>training_pipeline.run()
training_pipeline.view_statistics(magic=True)
training_pipeline.evaluate(magic=True)<jupyter_output><empty_output><jupyter_text>Inspect datasource<jupyter_code>datasources = repo.get_datasources()
datasource = datasources[0]
print(datasource)
df = datasource.sample_data()
df.head()
df.shape
df.columns<jupyter_output><empty_output><jupyter_text>Skip preprocessing with your next (warm-starting) pipeline  Clone first experiment and only change one hyper-parameter<jupyter_code>training_pipeline_2 = training_pipeline.copy('Experiment 2')
training_pipeline_2.add_trainer(TFFeedForwardTrainer(
    loss='binary_crossentropy',
    last_activation='sigmoid',
    output_units=1,
    metrics=['accuracy'],
    epochs=20))
training_pipeline_2.run()
training_pipeline_2.evaluate(magic=True)<jupyter_output><empty_output><jupyter_text>Post-training  Verify theres still only one datasource<jupyter_code>datasources = repo.get_datasources()
print(f"We have {len(datasources)} datasources")<jupyter_output><empty_output><jupyter_text>Compare pipelines<jupyter_code>repo.compare_training_runs()<jupyter_output><empty_output><jupyter_text>Distribute splitting/preprocessing easily<jupyter_code>training_pipeline_3 = repo.get_pipeline_by_name('Experiment 1').copy('Experiment 3')

# Define the processing backend
processing_backend = ProcessingDataFlowBackend(
    project=GCP_PROJECT,
    staging_location=os.path.join(GCP_BUCKET, 'dataflow_processing/staging'),
)


# Run processing step with that backend
training_pipeline_3.add_split(
    RandomSplit(split_map={'train': 0.7, 'eval': 0.2, 'test': 0.1}).with_backend(
        processing_backend)
)

training_pipeline_3.run(artifact_store=cloud_artifact_store)<jupyter_output><empty_output><jupyter_text>Easily train on the cloud<jupyter_code>training_pipeline_4 = training_pipeline.copy('Experiment 4')

# Add a trainer with a GCAIP backend
training_backend = SingleGPUTrainingGCAIPBackend(
    project=GCP_PROJECT,
    job_dir=TRAINING_JOB_DIR
)

training_pipeline_4.add_trainer(TFFeedForwardTrainer(
    loss='binary_crossentropy',
    last_activation='sigmoid',
    output_units=1,
    metrics=['accuracy'],
    epochs=20).with_backend(training_backend))

training_pipeline_4.run(artifact_store=cloud_artifact_store)<jupyter_output><empty_output><jupyter_text>Orchestrate pipeline whereever you like<jupyter_code>training_pipeline_5 = training_pipeline.copy('Experiment 5')

# Define the orchestrator backend
cloud_orchestrator_backend = OrchestratorGCPBackend(
    cloudsql_connection_name=CONNECTION_NAME,
    project=GCP_PROJECT,
    preemptible=True,  # reduce costs by using preemptible instances
    machine_type='n1-standard-4',
    gpu='nvidia-tesla-k80',
    gpu_count=1,
)

# Run the pipeline
training_pipeline_5.run(
    backend=cloud_orchestrator_backend,
    metadata_store=cloud_metadata_store,
    artifact_store=cloud_artifact_store,
)<jupyter_output><empty_output><jupyter_text>Add a deployer step with different integrations  Option 1: Deploy to Google Cloud AI Platform<jupyter_code>training_pipeline_6 = training_pipeline.copy('Experiment 6')
training_pipeline_6.add_deployment(
    GCAIPDeployer(
        project_id=GCP_PROJECT,
        model_name=MODEL_NAME,
    )
)

training_pipeline_6.run(artifact_store=cloud_artifact_store)<jupyter_output><empty_output><jupyter_text>Option 2: Deploy to Kubernetes via Cortex<jupyter_code>training_pipeline_7 = training_pipeline.copy('Experiment 7')

# Add cortex deployer
api_config = {
    "name": MODEL_NAME,
    "kind": "RealtimeAPI",
    "predictor": {
        "type": "tensorflow",
        "models": {"signature_key": "serving_default"}}
}
training_pipeline_7.add_deployment(
    CortexDeployer(
        env=CORTEX_ENV,
        api_config=api_config,
        predictor=TensorFlowPredictor,
    )
)

training_pipeline_7.run(artifact_store=cloud_artifact_store)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10661287998847428, 'fraction_numerical': 0.007779858809969745, 'mean_word_length': 7.234875444839858, 'pattern_counts': {'":': 6, '<': 71, '<?xml version=': 0, '>': 71, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '17', 'dir': 'jupyter-structured-clean-dedup', 'id': '53027', 'license': 'Apache-2.0', 'n_tokens_mistral': 2442, 'n_tokens_neox': 2424, 'n_words': 350, 'path': 'examples/backends/demo_modular.ipynb', 'repo_name': 'LpLegend/zenml', 'size': '6939'}