**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
)
            cluster_data['master_config']['image_uri'] = custom_image_url
            if not self.single_node:
                cluster_data['worker_config']['image_uri'] = custom_image_url
        cluster_data = self._build_gce_cluster_config(cluster_data)
        if self.single_node:
            self.properties["dataproc:dataproc.allow.zero.workers"] = "true"
        if self.properties:
            cluster_data['software_config']['properties'] = self.properties
        if self.optional_components:
            cluster_data['software_config']['optional_components'] = self.optional_components
        cluster_data = self._build_lifecycle_config(cluster_data)
        if self.init_actions_uris:
            init_actions_dict = [
                {'executable_file': uri, 'execution_timeout': self._get_init_action_timeout()}
                for uri in self.init_actions_uris
            ]
            cluster_data['initialization_actions'] = init_actions_dict
        if self.customer_managed_key:
            cluster_data['encryption_config'] = {'gce_pd_kms_key_name': self.customer_managed_key}
        if self.autoscaling_policy:
            cluster_data['autoscaling_config'] = {'policy_uri': self.autoscaling_policy}
        return cluster_data
    def make(self):
        """
        Helper method for easier migration.
        :return: Dict representing Dataproc cluster.
        """
        return self._build_cluster_data()
class DataprocCreateClusterOperator(BaseOperator):