
**Apache Airflow version**: 1.10.15

**OS**: Linux 5.4.109+

**Apache Airflow Provider versions**:
apache-airflow-backport-providers-apache-beam==2021.3.13
apache-airflow-backport-providers-cncf-kubernetes==2021.3.3
apache-airflow-backport-providers-google==2021.3.3

**Deployment**: Cloud Composer 1.16.6 (Google Cloud Managed Airflow Service)

**What happened**:

BigQueryCreateExternalTableOperator from the providers package ([airflow.providers.google.cloud.operators.bigquery](https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/operators/bigquery.py)) fails with correct _schema_object_ parameter.

**What you expected to happen**:

I expected the DAG to succesfully run, as I've previously tested it with the deprecated operator from the contrib package ([airflow.contrib.operators.bigquery_operator](https://github.com/apache/airflow/blob/5786dcdc392f7a2649f398353a0beebef01c428e/airflow/contrib/operators/bigquery_operator.py#L476)), using the same parameters.

Debbuging the DAG execution log, I saw the providers operator generated a wrong call to the Cloud Storage API: it mixed up the bucket and object parameters, according the stack trace bellow.

```
[2021-08-23 23:17:22,316] {taskinstance.py:1152} ERROR - 404 GET https://storage.googleapis.com/download/storage/v1/b/foo/bar/schema.json/o/mybucket?alt=media: Not Found: ('Request failed with status code', 404, 'Expected one of', <HTTPStatus.OK: 200>, <HTTPStatus.PARTIAL_CONTENT: 206>)
Traceback (most recent call last)
  File "/opt/python3.6/lib/python3.6/site-packages/google/cloud/storage/client.py", line 728, in download_blob_to_fil
    checksum=checksum
  File "/opt/python3.6/lib/python3.6/site-packages/google/cloud/storage/blob.py", line 956, in _do_downloa
    response = download.consume(transport, timeout=timeout
  File "/opt/python3.6/lib/python3.6/site-packages/google/resumable_media/requests/download.py", line 168, in consum
    self._process_response(result
  File "/opt/python3.6/lib/python3.6/site-packages/google/resumable_media/_download.py", line 186, in _process_respons
    response, _ACCEPTABLE_STATUS_CODES, self._get_status_cod
  File "/opt/python3.6/lib/python3.6/site-packages/google/resumable_media/_helpers.py", line 104, in require_status_cod
    *status_code
google.resumable_media.common.InvalidResponse: ('Request failed with status code', 404, 'Expected one of', <HTTPStatus.OK: 200>, <HTTPStatus.PARTIAL_CONTENT: 206>

During handling of the above exception, another exception occurred

Traceback (most recent call last)
  File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 985, in _run_raw_tas
    result = task_copy.execute(context=context
  File "/usr/local/lib/airflow/airflow/providers/google/cloud/operators/bigquery.py", line 1178, in execut
    schema_fields = json.loads(gcs_hook.download(self.bucket, self.schema_object)
  File "/usr/local/lib/airflow/airflow/providers/google/cloud/hooks/gcs.py", line 301, in downloa
    return blob.download_as_string(
  File "/opt/python3.6/lib/python3.6/site-packages/google/cloud/storage/blob.py", line 1391, in download_as_strin
    timeout=timeout
  File "/opt/python3.6/lib/python3.6/site-packages/google/cloud/storage/blob.py", line 1302, in download_as_byte
    checksum=checksum
  File "/opt/python3.6/lib/python3.6/site-packages/google/cloud/storage/client.py", line 731, in download_blob_to_fil
    _raise_from_invalid_response(exc
  File "/opt/python3.6/lib/python3.6/site-packages/google/cloud/storage/blob.py", line 3936, in _raise_from_invalid_respons
    raise exceptions.from_http_status(response.status_code, message, response=response
google.api_core.exceptions.NotFound: 404 GET https://storage.googleapis.com/download/storage/v1/b/foo/bar/schema.json/o/mybucket?alt=media: Not Found: ('Request failed with status code', 404, 'Expected one of', <HTTPStatus.OK: 200>, <HTTPStatus.PARTIAL_CONTENT: 206>
```
PS: the bucket (_mybucket_) and object path (_foo/bar/schema.json_) were masked for security reasons.

I believe the error appears on the [following](https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/operators/bigquery.py#L1183) line, although the bug itself is probably located on the [gcs_hook.download()](https://github.com/apache/airflow/blob/0264fea8c2024d7d3d64aa0ffa28a0cfa48839cd/airflow/providers/google/cloud/hooks/gcs.py#L266) method:

`schema_fields = json.loads(gcs_hook.download(self.bucket, self.schema_object))`

**How to reproduce it**:

Create a DAG using both operators and the same parameters, as the example bellow. The task using the contrib version of the operator should work, while the task using the providers version should fail.

```
from airflow.contrib.operators.bigquery_operator import BigQueryCreateExternalTableOperator as BQExtTabOptContrib
from airflow.providers.google.cloud.operators.bigquery import BigQueryCreateExternalTableOperator as BQExtTabOptProviders

#TODO: default args and DAG definition

create_landing_external_table_contrib = BQExtTabOptContrib(
    task_id='create_landing_external_table_contrib',
    bucket='mybucket',
    source_objects=['foo/bar/*.csv'],
    destination_project_dataset_table='project.dataset.table',
    schema_object='foo/bar/schema_file.json',
)

create_landing_external_table_providers = BQExtTabOptProviders(
    task_id='create_landing_external_table_providers',
    bucket='mybucket',
    source_objects=['foo/bar/*.csv'],
    destination_project_dataset_table='project.dataset.table',
    schema_object='foo/bar/schema_file.json',
)
```

**Anything else we need to know**:

The [*gcs_hook.download()*](https://github.com/apache/airflow/blob/0264fea8c2024d7d3d64aa0ffa28a0cfa48839cd/airflow/providers/google/cloud/hooks/gcs.py#L313) method is using the deprecated method _download_as_string()_ from the Cloud Storage API (https://googleapis.dev/python/storage/latest/blobs.html). It should be changed to _download_as_bytes()_.

Also, comparing the [providers version](https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/operators/bigquery.py#L1183) of the operator to the [contrib version](https://github.com/apache/airflow/blob/5786dcdc392f7a2649f398353a0beebef01c428e/airflow/contrib/operators/bigquery_operator.py#L621), I observed there is also a missing decode operation: `.decode("utf-8")`

**Are you willing to submit a PR?**

Yes
airflow/providers/google/cloud/utils/mlengine_operator_utils.py
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:
    .. code-block:: python
        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)
                squared_err = math.pow(classes - label, 2)
                return (err, squared_err)  # returns a tuple.
            return error_and_squared_error, ["err", "mse"]  # key order must match.
        def validate_err_and_count(summary):