**Description**
Currently, we can only pass custom Image name to  **DataprocClusterCreateOperator(),**
as the custom image expires after 60 days, we either need to update the image or we need to pass the expiration token.

Functionality is already available in **gcloud** and **REST**.

`gcloud dataproc clusters test_cluster ......
--image projects/<custom_image_project_id>/global/images/family/<family_name>
.......
`


**Use case / motivation**
The user should be able to pass either Custom Image or Custom Image family name,
so we don't have to update the image up on expiration or use expiration token.


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

**Related Issues**
None
airflow/providers/google/cloud/operators/dataproc.py
self.worker_disk_size = worker_disk_size
        self.zone = zone
        self.network_uri = network_uri
        self.subnetwork_uri = subnetwork_uri
        self.internal_ip_only = internal_ip_only
        self.tags = tags
        self.service_account = service_account
        self.service_account_scopes = service_account_scopes
        self.idle_delete_ttl = idle_delete_ttl
        self.auto_delete_time = auto_delete_time
        self.auto_delete_ttl = auto_delete_ttl
        self.customer_managed_key = customer_managed_key
        self.single_node = num_workers == 0
        if self.custom_image and self.image_version:
            raise ValueError("The custom_image and image_version can't be both set")
        if self.single_node and self.num_preemptible_workers > 0:
            raise ValueError("Single node cannot have preemptible workers.")
    def _get_init_action_timeout(self) -> dict:
        match = re.match(r"^(\d+)([sm])$", self.init_action_timeout)
        if match:
            val = float(match.group(1))
            if match.group(2) == "s":
                return {"seconds": int(val)}
            elif match.group(2) == "m":
                return {"seconds": int(timedelta(minutes=val).total_seconds())}
        raise AirflowException(
            "DataprocClusterCreateOperator init_action_timeout"
            " should be expressed in minutes or seconds. i.e. 10m, 30s"
        )
    def _build_gce_cluster_config(self, cluster_data):