**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
"""Scale, up or down, a cluster on Google Cloud Dataproc."""
        self.log.info("Scaling cluster: %s", self.cluster_name)
        scaling_cluster_data = self._build_scale_cluster_data()
        update_mask = ["config.worker_config.num_instances", "config.secondary_worker_config.num_instances"]
        hook = DataprocHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
        operation = hook.update_cluster(
            project_id=self.project_id,
            location=self.region,
            cluster_name=self.cluster_name,
            cluster=scaling_cluster_data,
            graceful_decommission_timeout=self._graceful_decommission_timeout_object,
            update_mask={'paths': update_mask},
        )
        operation.result()
        self.log.info("Cluster scaling finished")
class DataprocDeleteClusterOperator(BaseOperator):
    """
    Deletes a cluster in a project.
    :param project_id: Required. The ID of the Google Cloud project that the cluster belongs to (templated).
    :type project_id: str
    :param region: Required. The Cloud Dataproc region in which to handle the request (templated).
    :type region: str
    :param cluster_name: Required. The cluster name (templated).
    :type cluster_name: str