docs: def configure_pipeline(p, dataset, model_dir, output_csv, output_bq_table):
    """Configures a dataflow pipeline for batch prediction."""
  
    data = _util.get_sources_from_dataset(p, dataset, 'predict')
    if len(dataset.schema) == 2:
      output_schema = [
          {'name': 'image_url', 'type': 'STRING'},
          {'name': 'target', 'type': 'STRING'},
          {'name': 'predicted', 'type': 'STRING'},
          {'name': 'target_prob', 'type': 'FLOAT'},
          {'name': 'predicted_prob', 'type': 'FLOAT'},
      ]
    else:
      output_schema = [
          {'name': 'image_url', 'type': 'STRING'},
          {'name': 'predicted', 'type': 'STRING'},
          {'name': 'predicted_prob', 'type': 'FLOAT'},
      ]
    results = (data |
               'Load Images' >> beam.ParDo(LoadImagesDoFn()) |
               'Batch Inputs' >> beam.ParDo(EmitAsBatchDoFn(20)) |
               'Batch Predict' >> beam.ParDo(PredictBatchDoFn(model_dir)) |
               'Unbatch' >> beam.ParDo(UnbatchDoFn()) |
               'Process Results' >> beam.ParDo(ProcessResultsDoFn()))
  
    if output_csv is not None:
      schema_file = output_csv + '.schema.json'
      results_save = (results |
                      'Prepare For Output' >> beam.ParDo(MakeCsvLineDoFn()) |
                      'Write Csv Results' >> beam.io.textio.WriteToText(output_csv,
                                                                        shard_name_template=''))
      (results_save |
       'Sample One' >> beam.transforms.combiners.Sample.FixedSizeGlobally(1) |
       'Serialize Schema' >> beam.Map(lambda path: json.dumps(output_schema)) |
       'Write Schema' >> beam.io.textio.WriteToText(schema_file, shard_name_template=''))
  
    if output_bq_table is not None:
      # BigQuery sink takes schema in the form of 'field1:type1,field2:type2...'
      bq_schema_string = ','.join(x['name'] + ':' + x['type'] for x in output_schema)
      sink = beam.io.BigQuerySink(output_bq_table, schema=bq_schema_string,
                                  write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE)
      results | 'Write BQ Results' >> beam.io.Write(sink)