Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Samples - Moved secret application to the pipeline definition #536

Merged
merged 3 commits into from
Dec 14, 2018
Merged

Samples - Moved secret application to the pipeline definition #536

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
83feffd
Samples - Moved use_gcp_secret application to the pipeline definition
Ark-kun Dec 13, 2018
8a14392
Also changed the notebooks
Ark-kun Dec 14, 2018
7418fb5
Addressed the formatting feedback
Ark-kun Dec 14, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
Samples - Moved use_gcp_secret application to the pipeline definition
  • Loading branch information
@Ark-kun
Ark-kun committed Dec 13, 2018
commit 83feffd960c268f64b463bb81108ab392abe3cc0
16 changes: 8 additions & 8 deletions samples/kubeflow-tf/kubeflow-training-classification.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ def dataflow_tf_transform_op(train_data: 'GcsUri', evaluation_data: 'GcsUri', sc
'--output', transform_output,
],
file_outputs = {'transformed': '/output.txt'}
).apply(gcp.use_gcp_secret('user-gcp-sa'))
)


def kubeflow_tf_training_op(transformed_data_dir, schema: 'GcsUri[text/json]', learning_rate: float, hidden_layer_size: int, steps: int, target, preprocess_module: 'GcsUri[text/code/python]', training_output: 'GcsUri[Directory]', step_name='training'):
Expand All @@ -50,7 +50,7 @@ def kubeflow_tf_training_op(transformed_data_dir, schema: 'GcsUri[text/json]', l
'--job-dir', training_output,
],
file_outputs = {'train': '/output.txt'}
).apply(gcp.use_gcp_secret('user-gcp-sa'))
)

def dataflow_tf_predict_op(evaluation_data: 'GcsUri', schema: 'GcsUri[text/json]', target: str, model: 'TensorFlow model', predict_mode, project: 'GcpProject', prediction_output: 'GcsUri', step_name='prediction'):
return dsl.ContainerOp(
Expand All @@ -66,7 +66,7 @@ def dataflow_tf_predict_op(evaluation_data: 'GcsUri', schema: 'GcsUri[text/json]
'--output', prediction_output,
],
file_outputs = {'prediction': '/output.txt'}
).apply(gcp.use_gcp_secret('user-gcp-sa'))
)

def confusion_matrix_op(predictions, output, step_name='confusionmatrix'):
return dsl.ContainerOp(
Expand All @@ -76,7 +76,7 @@ def confusion_matrix_op(predictions, output, step_name='confusionmatrix'):
'--predictions', predictions,
'--output', output,
]
).apply(gcp.use_gcp_secret('user-gcp-sa'))
)

@dsl.pipeline(
name='Pipeline TFJob',
Expand All @@ -97,10 +97,10 @@ def kubeflow_training(output, project,
# TODO: use the argo job name as the workflow
workflow = '{{workflow.name}}'

preprocess = dataflow_tf_transform_op(train, evaluation, schema, project, preprocess_mode, '', '%s/%s/transformed' % (output, workflow))
training = kubeflow_tf_training_op(preprocess.output, schema, learning_rate, hidden_layer_size, steps, target, '', '%s/%s/train' % (output, workflow))
prediction = dataflow_tf_predict_op(evaluation, schema, target, training.output, predict_mode, project, '%s/%s/predict' % (output, workflow))
confusion_matrix = confusion_matrix_op(prediction.output, '%s/%s/confusionmatrix' % (output, workflow))
preprocess = dataflow_tf_transform_op(train, evaluation, schema, project, preprocess_mode, '', '%s/%s/transformed' % (output, workflow)).apply(gcp.use_gcp_secret('user-gcp-sa'))
training = kubeflow_tf_training_op(preprocess.output, schema, learning_rate, hidden_layer_size, steps, target, '', '%s/%s/train' % (output, workflow)).apply(gcp.use_gcp_secret('user-gcp-sa'))
prediction = dataflow_tf_predict_op(evaluation, schema, target, training.output, predict_mode, project, '%s/%s/predict' % (output, workflow)).apply(gcp.use_gcp_secret('user-gcp-sa'))
confusion_matrix = confusion_matrix_op(prediction.output, '%s/%s/confusionmatrix' % (output, workflow)).apply(gcp.use_gcp_secret('user-gcp-sa'))


if __name__ == '__main__':
Expand Down
26 changes: 13 additions & 13 deletions samples/tfx/taxi-cab-classification-pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ def dataflow_tf_data_validation_op(inference_data: 'GcsUri', validation_data: 'G
'schema': '/schema.txt',
'validation': '/output_validation_result.txt',
}
).apply(gcp.use_gcp_secret('user-gcp-sa'))
)

def dataflow_tf_transform_op(train_data: 'GcsUri', evaluation_data: 'GcsUri', schema: 'GcsUri[text/json]', project: 'GcpProject', preprocess_mode, preprocess_module: 'GcsUri[text/code/python]', transform_output: 'GcsUri[Directory]', step_name='preprocess'):
return dsl.ContainerOp(
Expand All @@ -51,7 +51,7 @@ def dataflow_tf_transform_op(train_data: 'GcsUri', evaluation_data: 'GcsUri', sc
'--output', '%s/{{workflow.name}}/transformed' % transform_output,
],
file_outputs = {'transformed': '/output.txt'}
).apply(gcp.use_gcp_secret('user-gcp-sa'))
)


def tf_train_op(transformed_data_dir, schema: 'GcsUri[text/json]', learning_rate: float, hidden_layer_size: int, steps: int, target: str, preprocess_module: 'GcsUri[text/code/python]', training_output: 'GcsUri[Directory]', step_name='training'):
Expand All @@ -69,7 +69,7 @@ def tf_train_op(transformed_data_dir, schema: 'GcsUri[text/json]', learning_rate
'--job-dir', '%s/{{workflow.name}}/train' % training_output,
],
file_outputs = {'train': '/output.txt'}
).apply(gcp.use_gcp_secret('user-gcp-sa'))
)

def dataflow_tf_model_analyze_op(model: 'TensorFlow model', evaluation_data: 'GcsUri', schema: 'GcsUri[text/json]', project: 'GcpProject', analyze_mode, analyze_slice_column, analysis_output: 'GcsUri', step_name='analysis'):
return dsl.ContainerOp(
Expand All @@ -85,7 +85,7 @@ def dataflow_tf_model_analyze_op(model: 'TensorFlow model', evaluation_data: 'Gc
'--output', '%s/{{workflow.name}}/analysis' % analysis_output,
],
file_outputs = {'analysis': '/output.txt'}
).apply(gcp.use_gcp_secret('user-gcp-sa'))
)


def dataflow_tf_predict_op(evaluation_data: 'GcsUri', schema: 'GcsUri[text/json]', target: str, model: 'TensorFlow model', predict_mode, project: 'GcpProject', prediction_output: 'GcsUri', step_name='prediction'):
Expand All @@ -102,7 +102,7 @@ def dataflow_tf_predict_op(evaluation_data: 'GcsUri', schema: 'GcsUri[text/json]
'--output', '%s/{{workflow.name}}/predict' % prediction_output,
],
file_outputs = {'prediction': '/output.txt'}
).apply(gcp.use_gcp_secret('user-gcp-sa'))
)


def confusion_matrix_op(predictions: 'GcsUri', output: 'GcsUri', step_name='confusion_matrix'):
Expand Down Expand Up @@ -157,18 +157,18 @@ def taxi_cab_classification(
analyze_slice_column='trip_start_hour'):

tf_server_name = 'taxi-cab-classification-model-{{workflow.name}}'
validation = dataflow_tf_data_validation_op(train, evaluation, column_names, key_columns, project, mode, output)
validation = dataflow_tf_data_validation_op(train, evaluation, column_names, key_columns, project, mode, output).apply(gcp.use_gcp_secret('user-gcp-sa'))
preprocess = dataflow_tf_transform_op(train, evaluation, validation.outputs['schema'],
project, mode, preprocess_module, output)
project, mode, preprocess_module, output).apply(gcp.use_gcp_secret('user-gcp-sa'))
training = tf_train_op(preprocess.output, validation.outputs['schema'], learning_rate,
hidden_layer_size, steps, 'tips', preprocess_module, output)
hidden_layer_size, steps, 'tips', preprocess_module, output).apply(gcp.use_gcp_secret('user-gcp-sa'))
analysis = dataflow_tf_model_analyze_op(training.output, evaluation,
validation.outputs['schema'], project, mode, analyze_slice_column, output)
validation.outputs['schema'], project, mode, analyze_slice_column, output).apply(gcp.use_gcp_secret('user-gcp-sa'))
Ark-kun marked this conversation as resolved.
Show resolved Hide resolved
prediction = dataflow_tf_predict_op(evaluation, validation.outputs['schema'], 'tips',
training.output, mode, project, output)
cm = confusion_matrix_op(prediction.output, output)
roc = roc_op(prediction.output, output)
deploy = kubeflow_deploy_op(training.output, tf_server_name)
training.output, mode, project, output).apply(gcp.use_gcp_secret('user-gcp-sa'))
cm = confusion_matrix_op(prediction.output, output).apply(gcp.use_gcp_secret('user-gcp-sa'))
roc = roc_op(prediction.output, output).apply(gcp.use_gcp_secret('user-gcp-sa'))
deploy = kubeflow_deploy_op(training.output, tf_server_name).apply(gcp.use_gcp_secret('user-gcp-sa'))

if __name__ == '__main__':
import kfp.compiler as compiler
Expand Down