Convert csv to feather file format - Download and upload from GCP bucket

README.md

Convert csv files to feather format files

Write and upload indicators file to GCP buckets

Install

Install python virtual env (or similar)

pip install virtualenv

Create python virtual env (or similar)

virtualenv -p python3.8 env

(Requires python >= 3.8)

Install dependencies

pip install -r requirements.txt

Run

Set environmental variables

• TEMPORARY_DOWNLOADS_PATH: Directory path where temporary downloaded files will be stored.
• BUCKET_NAME: GCP bucket name where script will search for and upload indicators files.
• GOOGLE_APPLICATION_CREDENTIALS (optional): Path to GCP credentials with read and write access to the bucket.

Run script

Example:

python main.py

main.py

import numpy as np
import pandas as pd
import os
import tqdm
import gc
import feather
from google.cloud import storage

PATH = os.getenv('TEMPORARY_DOWNLOADS_PATH', './downloads/')

bucket_name = os.getenv('BUCKET_NAME', 'modeler-non-prod')

os.makedirs(PATH, exist_ok=True)  # succeeds even if directory exists.

storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
blobs = storage_client.list_blobs(bucket_name)

for blob in blobs:
    if 'indicator' in blob.name and blob.name.endswith('.csv'):
        print('Indicator file found: ' + blob.name)

        filename = PATH + blob.name.replace('/', '_')
        feather_filename = os.path.splitext(filename)[0] + '.ftr'
        destination_blob_name = os.path.splitext(blob.name)[0] + '.ftr'

        destination_blob_exists = storage.Blob(bucket=bucket, name=destination_blob_name).exists(storage_client)
        if destination_blob_exists:
            print('Destination file already exists, skipping...')
            continue

        print('Downloading...')
        blob.download_to_filename(filename)

        print('Indicator file downloaded: ' + filename)
        print('Converting to feather file format...')

        df = pd.read_csv(filename)
        df.reset_index()
        df.to_feather(feather_filename)

        print('Indicator feather file created: ' + feather_filename)

        print('Uploading to bucket as \'' + destination_blob_name + '\'...')
        new_blob = bucket.blob(destination_blob_name)
        new_blob.upload_from_filename(feather_filename)
        print('File uploaded: ' + destination_blob_name)

        print('Deleting local files...')
        os.remove(filename)
        os.remove(feather_filename)

        print('Indicator file DONE! \n')

requirements.txt

numpy>=1.20.3
pandas
geopandas
Flask==2.1.0
tqdm
feather-format
google-api-python-client
google-cloud-storage