<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def batch_predict_async(dataset, model_dir, output_csv=None, output_bq_table=None, cloud=None):
 """Batch prediction with an offline model. Args: dataset: CsvDataSet or BigQueryDataSet for batch prediction input. Can contain either one column 'image_url', or two columns with another being 'label'. model_dir: The directory of a trained inception model. Can be local or GCS paths. output_csv: The output csv file for prediction results. If specified, it will also output a csv schema file with the name output_csv + '.schema.json'. output_bq_table: if specified, the output BigQuery table for prediction results. output_csv and output_bq_table can both be set. cloud: a DataFlow pipeline option dictionary such as {'num_workers': 3}. If anything but not None, it will run in cloud. Otherwise, it runs locally. If specified, it must include 'temp_location' with value being a GCS path, because cloud run requires a staging GCS directory. Raises: ValueError if both output_csv and output_bq_table are None, or if cloud is not None but it does not include 'temp_location'. Returns: A google.datalab.utils.Job object that can be used to query state from or wait. """

  with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    if cloud is None:
      return _local.Local.batch_predict(dataset, model_dir, output_csv, output_bq_table)

    if not isinstance(cloud, dict):
      cloud = {}
    return _cloud.Cloud.batch_predict(dataset, model_dir, output_csv, output_bq_table, cloud)