'Checks whether the file specified by file_name is stored in Google Cloud
Storage (GCS), if so, downloads the file and saves it locally. The full
path of the saved file will be returned. Otherwise the local file_name
will be returned immediately.
:param file_name: The full path of input file.
:type file_name: string
:return: The full path of local file.
:type: string'

def google_cloud_to_local(self, file_name):

    if (not file_name.startswith('gs://')):
        return file_name
    path_components = file_name[self.GCS_PREFIX_LENGTH:].split('/')
    if (path_components < 2):
        raise Exception('Invalid    Google    Cloud    Storage    (GCS)    object    path:    {}.'.format(file_name))
    bucket_id = path_components[0]
    object_id = '/'.join(path_components[1:])
    local_file = '/tmp/dataflow{}-{}'.format(str(uuid.uuid1())[:8], path_components[(-1)])
    file_size = self._gcs_hook.download(bucket_id, object_id, local_file)
    if (file_size > 0):
        return local_file
    raise Exception('Failed    to    download    Google    Cloud    Storage    GCS    object:    {}'.format(file_name))
