<jupyter_start><jupyter_text>_Clean & Store in BigQuery_<jupyter_code>import utils
import pandas as pd
from google.cloud import storage
from google.cloud import bigquery
import pandas_gbq
from pandas_gbq.gbq import TableCreationError, NotFoundException

%reload_ext autoreload
%autoreload 2
%matplotlib inline<jupyter_output><empty_output><jupyter_text>_May 29, 2020_<jupyter_code>#day = '2020-05-24'
#bucket_path = f'gs://thepanacealab_covid19twitter/dailies/{day}/{day}_clean-dataset.json'
def load_from_gcs(bucket_path):
    '''
    Takes path to file in Google Cloud Storage Bucket
    and returns a Pandas DataFrame
    '''
    df = pd.read_json(
        bucket_path,
        lines=True,
        dtype={
            'id_str': str,
            'in_reply_to_status_id_str': str,
            'quoted_status_id_str': str
        }
    )
    return df
#%%time
#df = load_from_gcs(bucket_path)
#%%time
#df = utils.load_data('2020-03-22')
#df.info()
def clean_for_parquet(df):
    '''
    Takes Panda DataFrame, cleans data into appropriate form for BigQuery.
    '''
    cols_of_interest = [
        'created_at',
        'id_str',
        'user',
        'lang',
        'full_text'
    ]
    df2clean = df.loc[:, cols_of_interest]
    df2clean['user_id_str'] = df2clean['user'].apply(lambda user: str(user['id_str']))
    df2clean.drop(labels = 'user', axis = 1, inplace = True)
    dfclean = df2clean[['created_at', 'id_str', 'user_id_str', 'lang', 'full_text']]
    return dfclean
#%%time
#dfclean = clean_for_parquet(df)
#assert len(dfclean) == len(df)
#dfclean.info()
def parquet_to_gcs(df, day):
    df.to_parquet(
        f'gs://thepanacealab_covid19twitter/dailies/{day}/{day}_tweets.parquet'
    )
    print('Dataframe uploaded to bucket as parquet file.')
def data_prep_wrapper(day):
    bucket_path = (
        f'gs://thepanacealab_covid19twitter/dailies/'
        + f'{day}/{day}_clean-dataset.json'
    )
    print(f'Loading data for {day}...')
    df = load_from_gcs(bucket_path)
    print(f'Cleaning data for {day}...')
    dfclean = clean_for_parquet(df)
    assert len(dfclean) == len(df)
    print(f'Converting to parquet file & storing in {day} bucket.')
    parquet_to_gcs(dfclean, day)
    print(f'{day} successfully converted and stored in Storage.\n')
#day = '2020-05-24'
#%%time
#data_prep_wrapper(day)
def list_json_dates(bucket_name='thepanacealab_covid19twitter'):
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    blobs = bucket.list_blobs(prefix='dailies/')
    json_files = [
        str(i).split(',')[1].strip() for i in blobs 
        if str(i).split(',')[1].endswith('.json')
    ]
    json_dates = [i.split('/')[1] for i in json_files]
    return json_dates
def list_parquet_dates(bucket_name='thepanacealab_covid19twitter'):
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    blobs = bucket.list_blobs(prefix='dailies/')
    parquet_files = [
        str(i).split(',')[1].strip() for i in blobs 
        if str(i).split(',')[1].endswith('.parquet')
    ]
    parquet_dates = [
        i.split('/')[1] for i in parquet_files
    ]
    return parquet_dates

def need_parquet_dates():
    json_dates = list_json_dates()
    parquet_dates = list_parquet_dates()
    need_parquet = sorted(list(set(json_dates) - set(parquet_dates)))
    return need_parquet[::-1]
need_parquet_dates()
json_dates = list_json_dates()
parquet_dates = list_parquet_dates()
print(parquet_dates)
print(json_dates)
need_parquet = sorted(list(set(json_dates) - set(parquet_dates)))
print(need_parquet)
%%time

# time with 3 most recent days
for day in need_parquet[-3:]:
    data_prep_wrapper(day)
#df.info()
#def bq_create_table(df, day):
#    '''
#    Takes Pandas DataFrame, and it's associated date, and stores in BigQuery table
#   '''
#   bigquery_client = bigquery.Client()
#    dataset_ref = bigquery_client.dataset('twitter_dailies')
#    # BigQuery only accepts underscores for table names
#    bq_day = day.replace('-', '_')
#    # Prepares a reference to the table
#    table_ref = dataset_ref.table(bq_day)
#
#    try:
#        found = bigquery_client.get_table(table_ref)
#        return found
#    except Exception as e:
#        table_ref = dataset_ref.table(bq_day)
#        job = bigquery_client.load_table_from_dataframe(df, table_ref, location='US')
        # waits for table load to complete
#        job.result()
#        print(
#            f'Loaded dataframe from {day} with {job.output_rows} observations into',
#            + f' {table_ref.path}'
#        )
def bq_create_table(df, day):
    '''
    Takes Pandas DataFrame, and it's associated date, and stores in BigQuery table
    '''
    # BigQuery only accepts underscores for table names
    bq_day = day.replace('-', '_')
    # prepares reference to the table
    try:
        table_schema = [
            {'name': 'created_at', 'type': 'TIMESTAMP'},
            {'name': 'id_str', 'type': 'STRING'},
            {'name': 'user_id_str', 'type': 'STRING'},
            {'name': 'lang', 'type': 'STRING'},
            {'name': 'full_text', 'type': 'STRING'}
        ]
        df.to_gbq(
            project_id='covid-disinfo-detect',
            destination_table=f'twitter_dailies.{bq_day}',
            if_exists='fail'
        )
    except TableCreationError as e:
        pass
#def bq_create_table(df, day):
#    '''
#    Takes Pandas DataFrame, and it's associated date, and stores in BigQuery table
#    '''
#    client = bigquery.Client()
#    dataset_ref = client.dataset('twitter_dailies')
#    # BigQuery only accepts underscores for table names
#    bq_day = day.replace('-', '_')
#    # prepares reference to the table
#    table_ref = dataset_ref.table(bq_day)
#    try:
#        found = client.get_table(table_ref)
#        print(found)
#    except NotFoundException as e:
#        job_config = bigquery.LoadJobConfig(schema=[
#            bigquery.SchemaField('created_at', 'TIMESTAMP'),
#            bigquery.SchemaField('id_str', 'STRING'),
#            bigquery.SchemaField('user_id_str', 'STRING'),
#            bigquery.SchemaField('lang', 'STRING'),
#            bigquery.SchemaField('full_text', 'STRING')
#        ])
#        job = client.load_table_from_dataframe(
#            df, table_ref, job_config=job_config
#        )
#        job.result()
#        print("Loaded dataframe to {}".format(table_ref.path))
#%%time
#bq_create_table(dfclean, day)
def data_ingestion_bigquery(day):
    bucket_path = (
        f'gs://thepanacealab_covid19twitter/dailies/{day}/'
        + f'{day}_clean-dataset.json'
    )
    print(f'Loading data for {day}...')
    df = load_from_gcs(bucket_path)
    print(f'Loaded data for {day}...')
    dfclean = clean_for_bigquery(df)
    print(f'Cleaned data for BigQuery...')
    assert len(dfclean) == len(df)
    bq_create_table(dfclean, day)
day = '2020-05-25'
#day_list = ['2020-05-25', '2020-05-24', '2020-05-23']
#%%time
#for day in day_list:
#    print(day)
#%%time
#for day in day_list:
#    data_ingestion_bigquery(day)
%%time
data_ingestion_bigquery(day)
%%time
#utils.download_json('2020-05-19')
sb_client = storage.Client()
buckets = sb_client.list_buckets()

print("Buckets in {}:".format(sb_client.project))
for item in buckets:
    print("\t" + item.name)
from datetime import timedelta, date

def daterange(date1, date2):
    for n in range(int((date2 - date1).days)+1):
        yield date1 + timedelta(n)
        
start_dt = date(2015, 12, 20)
end_dt = date(2016, 1, 11)
for dt in daterange(start_dt, end_dt):
    print(dt.strftime("%Y-%m-%d"))
def blob_exists(day, bucket_name='thepanacealab_covid19twitter'):
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    blob = bucket.blob(f'dailies/{day}/{day}_clean-dataset.json')
    return blob.exists()
def list_blobs_bucket(bucket_name='thepanacealab_covid19twitter'):
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    blobs = list(bucket.list_blobs(prefix='dailies/', suff))
    return blobs
list_blobs_bucket()
blob_exists('2020-05-21')
def bigquery_tables_list():
    # gather list of tables in bigquery
    client = bigquery.Client()
    dataset = client.get_dataset('twitter_dailies')
    tables = list(client.list_tables(dataset))
    return [table.table_id for table in tables]
bq_tables = bigquery_tables_list()
def list_blobs_bucket(bucket_name='thepanacealab_covid19twitter'):
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    blobs = bucket.list_blobs(prefix='dailies/')
    json_files = [
        str(i).split(',')[1].strip() for i in blobs if str(i).split(',')[1].endswith('.json')
    ]
    storage_dates = [i.split('/')[1].replace('-', '_') for i in json_files]
    return json_files, storage_dates
json_files, storage_dates = list_blobs_bucket()
bigquery_tables_list()
storage_dates
notinbq = sorted(list(set(storage_dates) - set(bq_tables)))
notinbq
for day in notinbq[:2]:
    day = day.replace('_', '-')
    utils.download_json(day)
    print(f'Downloaded data from {day}..')
def load_data(date, chunksize=50000):
    cols_interest = [
        'created_at',
        'id_str',
        'user',
        'lang',
        'full_text'
    ]
    
    chunks = pd.read_json(
        f'playground_data/{date}_clean-dataset.json',
        lines=True,
        chunksize=chunksize,
        dtype={
            'id_str': str,
            'in_reply_to_status_id_str': str,
            'quoted_status_id_str': str
        }
    )
    
    df = pd.concat(chunk for chunk in chunks)
    print('Loaded data...\n')
    return df[cols_interest]


def clean_data(df):
    df2clean = df.copy()
    df2clean['user_id_str'] = df2clean['user'].apply(lambda user: str(user['id_str']))
    df2clean.drop(labels = 'user', axis = 1, inplace = True)
    dfclean = df2clean[['created_at', 'id_str', 'user_id_str', 'lang', 'full_text']]
    print('Cleaned data...\n')
    return dfclean


def data_setup(date):
    df = clean_data(load_data(date))
    return df


def load_bigquery(df, date):
    client = bigquery.Client(location='US')
    dataset = client.get_dataset('twitter_dailies')
    bq_date = date.replace('-', '_')    # need to slightly change date string
    table_ref = dataset.table(bq_date)
    job = client.load_table_from_dataframe(df, table_ref, location='US')
    job.result()
    print(f'Loaded {job.output_rows} rows from dataframe to {table_ref.path}\n')
    
    
def data_bq_wrapper():
    date = input('What is the date of the data that you would like to store in BigQuery?\n')
    df = data_setup(date)
    load_bigquery(df, date)
    print(f'Data from {date} successfully stored in BigQuery.\n')
%%time
data_bq_wrapper()<jupyter_output>What is the date of the data that you would like to store in BigQuery?
 2020-05-19
{'fraction_non_alphanumeric': 0.13806870937790158, 'fraction_numerical': 0.014484679665738161, 'mean_word_length': 3.4161541615416153, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '282718', 'license': 'MIT', 'n_tokens_mistral': 3864, 'n_tokens_neox': 3739, 'n_words': 768, 'path': 'experiments/playground_nbs/clean-store-bigquery.ipynb', 'repo_name': 'earny-joe/CvDisinfo-Detect', 'size': '10770'}