human
Complete this python program:
def create_evaluate_ops(task_prefix, data_format, input_paths, prediction_path, metric_fn_and_keys, validate_fn, batch_prediction_job_id=None, project_id=None, region=None, dataflow_options=None, model_uri=None, model_name=None, version_name=None, dag=None):
assistant
#Here is the Python code you requested.

    """Creates Operators needed for model evaluation and returns.
    It gets prediction over inputs via Cloud ML Engine BatchPrediction API by
    calling CloudMLBatchPredictionOperator, then summarize and validate
    the result via Cloud Dataflow using DataFlowPythonOperator.
    For details and pricing about Batch prediction, please refer to the website
    https://cloud.google.com/ml-engine/docs/how-tos/batch-predict
    and for Cloud Dataflow, https://cloud.google.com/dataflow/docs/
    It returns three chained operators for prediction, summary, and validation,
    named as <prefix>-prediction, <prefix>-summary, and <prefix>-validation,
    respectively.
    (<prefix> should contain only alphanumeric characters or hyphen.)
    The upstream and downstream can be set accordingly like:
    pred, _, val = create_evaluate_ops(...)
    pred.set_upstream(upstream_op)
    downstream_op.set_upstream(val)
    Callers will provide two python callables, metric_fn and validate_fn, in
    order to customize the evaluation behavior as they wish.
    - metric_fn receives a dictionary per instance derived from json in the
    batch prediction result. The keys might vary depending on the model.
    It should return a tuple of metrics.
    - validation_fn receives a dictionary of the averaged metrics that metric_fn
    generated over all instances.
    The key/value of the dictionary matches to what\s given by
    metric_fn_and_keys arg.
    The dictionary contains an additional metric, \count\ to represent the
    total number of instances received for evaluation.
    The function would raise an exception to mark the task as failed, in a
    case the validation result is not okay to proceed (i.e. to set the trained
    version as default).
    Typical examples are like this:
    def get_metric_fn_and_keys():
    import math  # imports should be outside of the metric_fn below.
    def error_and_squared_error(inst):
    label = float(inst[\input_label\])
    classes = float(inst[\classes\])  # 0 or 1
    err = abs(classes-label)
..."""
    if (not re.match('^[a-zA-Z][-A-Za-z0-9]*$', task_prefix)):
        raise AirflowException(('Malformed    task_id    for    DataFlowPythonOperator    (only    alphanumeric    and    hyphens    are    allowed    but    got:    ' + task_prefix))
    (metric_fn, metric_keys) = metric_fn_and_keys
    if (not callable(metric_fn)):
        raise AirflowException('`metric_fn`    param    must    be    callable.')
    if (not callable(validate_fn)):
        raise AirflowException('`validate_fn`    param    must    be    callable.')
    if ((dag is not None) and (dag.default_args is not None)):
        default_args = dag.default_args
        project_id = (project_id or default_args.get('project_id'))
        region = (region or default_args.get('region'))
        model_name = (model_name or default_args.get('model_name'))
        version_name = (version_name or default_args.get('version_name'))
        dataflow_options = (dataflow_options or default_args.get('dataflow_default_options'))
    evaluate_prediction = CloudMLBatchPredictionOperator(task_id=(task_prefix + '-prediction'), project_id=project_id, job_id=batch_prediction_job_id, region=region, data_format=data_format, input_paths=input_paths, output_path=prediction_path, uri=model_uri, model_name=model_name, version_name=version_name, dag=dag)
    metric_fn_encoded = base64.b64encode(dill.dumps(metric_fn, recurse=True))
    evaluate_summary = DataFlowPythonOperator(task_id=(task_prefix + '-summary'), py_options=['-m'], py_file='airflow.contrib.operators.cloudml_prediction_summary', dataflow_default_options=dataflow_options, options={'prediction_path': prediction_path, 'metric_fn_encoded': metric_fn_encoded, 'metric_keys': ','.join(metric_keys)}, dag=dag)
    evaluate_summary.set_upstream(evaluate_prediction)
    def apply_validate_fn(*args, **kwargs):
        prediction_path = kwargs['templates_dict']['prediction_path']
        (scheme, bucket, obj, _, _) = urlsplit(prediction_path)
        if ((scheme != 'gs') or (not bucket) or (not obj)):
            raise...