<jupyter_start><jupyter_text>Analisis exploratorio inicial de datos de CAF 2. Proceso ETL- Carga de datos de parquet a pandas- Limpieza inicial de los datos- Tratamiento de datos categoricos- Tratamiento del campo fecha- Convertir el campo indice en fecha- Comprobacion de la frequencia de muestreo- Ajuste del dataset a la frequencia de muestreo V2- Deteccion del tren parado- Pendientes- Datos climatologicos<jupyter_code>!pip install git+https://github.com/maguelo/cafcoding.git#egg=cafcoding
import pyarrow.parquet as pq
import pandas as pd
import sys
import colorlog

from google.colab import drive
drive.mount('/content/drive')
from cafcoding.tools import aws
from cafcoding.tools import meteo
from cafcoding.tools import etl
import cafcoding.tools.log as log
from cafcoding.stages import etl_step
import logging
logger = log.get_logger()
#logger.setLevel(logging.INFO)<jupyter_output><empty_output><jupyter_text>Cargamos la sesion de Amazon<jupyter_code>amazon = aws.AWS("/content/drive/MyDrive/CAFcoding/config/awscli.ini")

# Activamos las credenciales
amazon.set_credentials_in_env()

for filename in amazon.ls('cafcodingdatos/raw'):
    print (filename)<jupyter_output>cafcodingdatos/raw/Unsaved
cafcodingdatos/raw/_SUCCESS
cafcodingdatos/raw/date=2020-10-01
cafcodingdatos/raw/date=2020-10-02
cafcodingdatos/raw/date=2020-10-03
cafcodingdatos/raw/date=2020-10-04
cafcodingdatos/raw/date=2020-10-05
cafcodingdatos/raw/date=2020-10-06
cafcodingdatos/raw/date=2020-10-07
cafcodingdatos/raw/date=2020-10-08
cafcodingdatos/raw/date=2020-10-09
cafcodingdatos/raw/date=2020-10-10
cafcodingdatos/raw/date=2020-10-11
cafcodingdatos/raw/date=2020-10-12
cafcodingdatos/raw/date=2020-10-13
cafcodingdatos/raw/date=2020-10-14
cafcodingdatos/raw/date=2020-10-15
cafcodingdatos/raw/date=2020-10-16
cafcodingdatos/raw/date=2020-10-17
cafcodingdatos/raw/date=2020-10-18
cafcodingdatos/raw/date=2020-10-19
cafcodingdatos/raw/date=2020-10-20
cafcodingdatos/raw/date=2020-10-21
cafcodingdatos/raw/date=2020-10-22
cafcodingdatos/raw/date=2020-10-23
cafcodingdatos/raw/date=2020-10-24
cafcodingdatos/raw/date=2020-10-25
cafcodingdatos/raw/date=2020-10-26
cafcodingdatos/raw/date=2020-10-27
[...]<jupyter_text>Cargamos el dataset<jupyter_code>meteo_path = r'drive/MyDrive/CAFcoding/notebook/utils/meteo_data'


parquet_csv_list =[]
GENERATE_TRAIN = True
GENERATE_VAL =False
GENERATE_TEST =True
NUMBER_DAYS = 1

PATH_TEMPLATE='s3://cafcodingdatos/targets_mirror/{step}/2020-{month:02d}-{day:02d}-etl-{version}.csv'

if GENERATE_TRAIN:
    STEP = 'train'
    MONTH = 10
    NUMBER_DAYS = 31
    START_DATE = 1 
    parquet_paths_s3 = ['cafcodingdatos/raw/date=2020-10-{0:02d}'.format(day) for day in range(START_DATE,NUMBER_DAYS+1)]
    csv_path  = [PATH_TEMPLATE.format(step=STEP,month=MONTH, day=day,version=etl_step.ETL_VERSION) for day in range(START_DATE,NUMBER_DAYS+1)]


    parquet_csv_list += list(zip(parquet_paths_s3,csv_path))

if GENERATE_VAL:
    STEP = 'val'
    MONTH = 11
    #NUMBER_DAYS = 10
    START_DATE = 1
    parquet_paths_s3 = ['cafcodingdatos/raw/date=2020-11-{0:02d}'.format(day) for day in range(START_DATE,NUMBER_DAYS+1)]
    csv_path  = [PATH_TEMPLATE.format(step=STEP,month=MONTH, day=day,version=etl_step.ETL_VERSION) for day in range(START_DATE,NUMBER_DAYS+1)]

    parquet_csv_list += list(zip(parquet_paths_s3,csv_path))

if GENERATE_TEST:
    STEP = 'test'
    MONTH = 11
    NUMBER_DAYS = 30
    START_DATE = 11
    parquet_paths_s3 = ['cafcodingdatos/raw/date=2020-11-{0:02d}'.format(day) for day in range(START_DATE,NUMBER_DAYS+1)]
    csv_path  = [PATH_TEMPLATE.format(step=STEP,month=MONTH, day=day,version=etl_step.ETL_VERSION) for day in range(START_DATE,NUMBER_DAYS+1)]

    parquet_csv_list += list(zip(parquet_paths_s3,csv_path))<jupyter_output><empty_output><jupyter_text>Funciones personalizadasDe esta forma se pueden probar funciones sin recargar el notebook<jupyter_code># def delete_stopped_train(df):
#     print ("test")
#     return df

# etl_step.delete_stopped_train = delete_stopped_train<jupyter_output><empty_output><jupyter_text>Cargamos datos desde S3<jupyter_code>df_meteo = meteo.load_meteo_dataframe(meteo_path)

df_list = []
for parquet_file, csv_file in parquet_csv_list:
    df_day_list=[]
    for filename in amazon.ls(parquet_file):
        if filename.endswith('.parquet'):
            try:
                print (filename)
                dataset = pq.ParquetDataset(filename, filesystem=amazon.fs)
                table = dataset.read()
                df = table.to_pandas() 
                df_day_list.append(df)
            except Exception as exc:
                logger.error("Se ha producido un error cargando el fichero de Parquet: {}",parquet_file)
                logger.error("{}",str(exc))
            
    
    df_day = pd.concat(df_day_list)
    try:
        df_day = etl_step.process_etl(df_day, df_meteo.copy(), del_stopped_train=True)
        df_day.to_csv(csv_file, index=True, index_label="ts_date_index")
    except Exception as exc:
        logger.error("Se ha producido un error al procesar el dataset: {} ",parquet_file)
        logger.error("{}",str(exc))
    df_list.append(df_day)




df = pd.concat(df_list)
df.head()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1113658070678128, 'fraction_numerical': 0.05425023877745941, 'mean_word_length': 4.691304347826087, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 20, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '425855', 'license': 'Apache-2.0', 'n_tokens_mistral': 2252, 'n_tokens_neox': 1995, 'n_words': 326, 'path': 'documentation/6_2_ETL_v2_pipeline.ipynb', 'repo_name': 'maguelo/cafcoding', 'size': '5235'}