<jupyter_start><jupyter_text>Use Pandas for ETL Now it’s time to write some simples ETL jobs for data analysis. Our scope is to create a fact table (denormalized) in our presentation area.In this notebook we will discuss which process steps (divided into extract, transform and load) we have to do to clean the source data, aggregate the records and, finally, load our records in our Document Store. Overview of our ETL steps: ![picture](https://drive.google.com/uc?id=1h60hvtzWmZYHJyuOaONpYiyNLmsQTlje)   Load and extract the source file First of all we need to load raw data (from CSV files) into our environment.<jupyter_code>from google.colab import files

uploaded = files.upload()
for fn in uploaded.keys():
  print('User uploaded file "{name}" with length {length} bytes'.format(
      name=fn, length=len(uploaded[fn])))<jupyter_output>User uploaded file "ds_project_details_full.csv" with length 2019128 bytes<jupyter_text>Most of our ETL code revolve around using the following functions:- drop_duplicates- dropna- replace / fillna- df[df['column'] != value]: filtering- apply: transform, or adding new column- merge: SQL like inner, left, or right join- groupby- read_csv / to_csvFunctions like drop_duplicates and drop_na are nice abstractions and save tens of SQL statements.And replace / fillna is a typical step that to manipulate the data array.All these features are available from pandas.<jupyter_code>import pandas as pd
import io
ds_project_details_full = pd.read_csv('/content/ds_project_details_full.csv')
pd.read_sql("select campo, count(*) from tabella group by campo")
ds_project_details_full = pd.read_csv(io.BytesIO(uploaded['ds_project_details_full.csv']))

ds_project_details_full.info()<jupyter_output><class 'pandas.core.frame.DataFrame'>
RangeIndex: 4600 entries, 0 to 4599
Data columns (total 24 columns):
 #   Column                Non-Null Count  Dtype  
---  ------                --------------  -----  
 0   Unnamed: 0            4600 non-null   int64  
 1   bullet_point          1 non-null      object 
 2   category              4600 non-null   object 
 3   category_url          4600 non-null   object 
 4   clickthrough_url      4600 non-null   object 
 5   close_date            4599 non-null   object 
 6   currency              4600 non-null   object 
 7   funds_raised_amount   4600 non-null   int64  
 8   funds_raised_percent  4600 non-null   float64
 9   image_url             4600 non-null   object 
 10  is_indemand           4600 non-null   bool   
 11  is_pre_launch         4600 non-null   bool   
 12  offered_by            0 non-null      float64
 13  open_date             4599 non-null   object 
 14  perk_goal_percentage  1 non-null      float64
 15  perks_claimed        [...]<jupyter_text>Data processing is often exploratory.We need to see the shape of the data, and write our next line of code based on our previous output. So the process is iterative.One tool that Python + Pandas comes in handy is Jupyter Notebook or Google Colab.<jupyter_code>ds_project_details_full.head()
number_of_records = ds_project_details_full.shape[0]
print(f"Number of records loaded {number_of_records}")<jupyter_output>Number of records loaded 4600<jupyter_text>Transform After loading the raw data, let's go do the initial cleaning tasks.Since we want to upload the data to MongoDB, we should immediately add a unique identifier (_id on MongoDB).The operations we will do are to create our staging table **st_projects** where:- we do not allow **duplicates**- we select only the **necessary columns**- remove **anomalous records**<jupyter_code># Add the id
ds_project_details_full['_id'] = ds_project_details_full['project_id']
# Remove duplicates
ds_project_no_duplicates = ds_project_details_full.drop_duplicates(subset=['title'])
ds_project_no_duplicates = ds_project_no_duplicates.drop_duplicates(subset=['tagline'])
number_of_records_without = ds_project_no_duplicates.shape[0]
print(f"-- Number of records without duplicates {number_of_records_without}")
ds_project_no_duplicates.info()
# Select only some features
ds_project_features = ds_project_no_duplicates[['_id', 'project_id', 'title',
                                                'tags', 'tagline', 'open_date', 'funds_raised_amount',
                                                'funds_raised_percent', 'currency', 'close_date', 'category']]
ds_project_features.head()
# Remove noise
ds_project_cleaned = ds_project_features[(ds_project_features['funds_raised_percent'] > 0) & (ds_project_features['funds_raised_percent'] < 1000)]
# Remove null values in title
ds_project_cleaned = ds_project_cleaned[ds_project_cleaned.tagline.notnull()]
number_of_records_without_noise = ds_project_cleaned.shape[0]
print(f"-- Number of records without noise {number_of_records_without_noise}")<jupyter_output>-- Number of records without noise 4559<jupyter_text>Load data in MongoDB Now the records are ready, following a Big Data approach:- we load the raw, raw data on a table with all the source data (**sc_projects**)- load the clean data in the staging table **st_projects**For the connection to MongoDB we will use the **pymongo** library.<jupyter_code>!pip install pymongo
import pymongo
import json
from pymongo import UpdateOne
client = pymongo.MongoClient("mongodb://xxxxx:xxxx@xxxxx:27017,xxxxx:27017,xxxx:27017/myFirstDatabase?ssl=true&replicaSet=atlas-14k1wg-shard-0&authSource=admin&retryWrites=true&w=majority")
db = client.indiegogo<jupyter_output><empty_output><jupyter_text>The data on MongoDB is in bson (**binary json**) format.We convert our dataframe pandas in json and create the list of update or insert on our collection.<jupyter_code>records = json.loads(ds_project_details_full.T.to_json()).values()
upserts=[UpdateOne({'_id':x['_id']}, {'$setOnInsert':x}, upsert=True) for x in records]
db.sc_project.bulk_write(upserts)
records = json.loads(ds_project_cleaned.T.to_json()).values()
upserts=[UpdateOne({'_id':x['_id']}, {'$setOnInsert':x}, upsert=True) for x in records]
db.st_project_cleaned.bulk_write(upserts)<jupyter_output><empty_output><jupyter_text>Extract and load image details We perform the same work now on the list of concepts extracted with the API from the images,<jupyter_code>from google.colab import files
uploaded = files.upload()
ds_img_details_full = pd.read_csv(io.BytesIO(uploaded['ds_img_details_full.csv']))
ds_img_details_full.info()
number_of_records = ds_img_details_full.shape[0]
print(f"Number of records loaded {number_of_records}")
ds_img_details_full.info()
ds_img_details_full.head()

records = json.loads(ds_img_details_full.T.to_json()).values()
db.sc_images.inser_many(upserts)

ds_images_aggregate.head()
ds_img_details_full['concepts'] = ds_img_details_full. \
  apply(lambda row: {'name': row['name'], 'value': row['value']}, axis=1)
ds_img_details_full.head()
ds_images_aggregate = ds_img_details_full.groupby('project_id')['concepts'].apply(list).reset_index(name="concepts")
ds_images_aggregate.head()<jupyter_output><empty_output><jupyter_text>Document databases also admit complex data types, so we go to load our records, where we have an array of concepts.<jupyter_code>ds_images_aggregate.info()
ds_images_aggregate['_id'] = ds_images_aggregate['project_id']
records = json.loads(ds_images_aggregate.T.to_json()).values()
upserts=[UpdateOne({'_id':x['_id']}, {'$setOnInsert':x}, upsert=True) for x in records]
db.st_concepts.bulk_write(upserts)<jupyter_output><empty_output><jupyter_text>Data to presentation layer Let's now build the final fact table: the goal is to create a denormalized table ready for analysis.<jupyter_code>client = pymongo.MongoClient("mongodb://xxxx:xx@xxxx:27017,xxxx:27017,xxx:27017/myFirstDatabase?ssl=true&replicaSet=atlas-14k1wg-shard-0&authSource=admin&retryWrites=true&w=majority")
db = client.indiegogo
st_project_cleaned = db.st_project_cleaned
st_concepts = db.st_concepts
st_project_cleaned.count_documents({})
st_concepts.count_documents({})
db.collection.find({}).forEach(function(x) {
    t = db.collection2.findOne({chiave: x.chiave})
})
### Example MongoDB -- NOT RUN!!!!
result = db.sc_images.aggregate([
    {
        '$match': {
            'value': {
                '$gt': 0.95
            }
        }
    }, {
        '$group': {
            '_id': '$project_id', 
            'count': {
                '$sum': 1
            }
        }
    }, {
        '$out': 'st_after_aggregate'
    }
])
# Join collections
df_concepts =  pd.DataFrame(list(st_concepts.find({}))).drop(columns=['_id'])
df_projects_cleaned =  pd.DataFrame(list(st_project_cleaned.find({})))

df_projects_cleaned.info()
df_ft_projects = df_projects_cleaned.merge(df_concepts, on='project_id', how='left')
df_ft_projects.head()
df_ft_projects.info()
records = json.loads(df_ft_projects.T.to_json()).values()
upserts=[UpdateOne({'_id':x['_id']}, {'$setOnInsert':x}, upsert=True) for x in records]
db.ft_projects.bulk_write(upserts)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10632119723028814, 'fraction_numerical': 0.018539200357382175, 'mean_word_length': 4.736707238949392, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 35, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '308078', 'license': 'MIT', 'n_tokens_mistral': 3012, 'n_tokens_neox': 2850, 'n_words': 740, 'path': 'notebooks/3c_Use_Pandas_for_ETL_Lession_2021_06_26.ipynb', 'repo_name': 'mauropelucchi/unimib_masterbi_2021', 'size': '8954'}