{"id":13138,"date":"2022-02-04T10:55:14","date_gmt":"2022-02-04T10:55:14","guid":{"rendered":"https:\/\/bluetab.net\/?p=13138"},"modified":"2022-02-04T10:55:14","modified_gmt":"2022-02-04T10:55:14","slug":"mi-experiencia-en-el-mundo-de-big-data-parte-2","status":"publish","type":"post","link":"https:\/\/bluetab.mx\/en\/2022\/02\/mi-experiencia-en-el-mundo-de-big-data-parte-2\/","title":{"rendered":"Mi experiencia en el mundo de Big Data – Parte II"},"content":{"rendered":"

Mi experiencia en el mundo de Big Data – Parte II<\/h1>\n
\"\"<\/a><\/figure>\n

David Emmanuel Reyes N\u00fa\u00f1ez<\/a><\/h4>\n

Senior Data Engineer<\/p>\n

\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\tShare on twitter
\n\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\tShare on linkedin<\/p>\n

En la entrega anterior (adjunto) creamos los scripts para enlistar y descargar archivos desde Google Drive hacia nuestro filesystem local.<\/p>\n

En esta entrega continuaremos con el c\u00f3digo de la funci\u00f3n processDriveFiles.py y crearemos los scripts para hacer la carga de archivos hacia Google Cloud<\/p>\n

La funcionalidad de este script es procesar los archivos listados en nuestro archivo parameters.csv, los cuales tengan el par\u00e1mero Status con valor 1, recordemos que esto le indica a nuestro programa si el archivo se descargar\u00e1 y procesar\u00e1 o no.<\/p>\n

A continuaci\u00f3n, el c\u00f3digo b\u00e1sico de esta funci\u00f3n. Para nuestro ejemplo solo incluiremos archivos con extensi\u00f3n csv<\/strong> y separados por pipes \u201c|\u201d.<\/strong><\/p>\n

En pasos anteriores ya descargamos nuestro archivo al servidor local, el paso siguiente ser\u00e1 ingestarlo en Big query y subir el archivo a nuestro proyecto de GCP.<\/p>\n

El siguiente c\u00f3digo se encarga de validar el archivo e ingestarlo hacia nuestro destino definido.<\/p>\n

#Validamos que el tamaño del archivo sea mayor a 0 para poder cargarlo al destino definido en el archivo de configuración, en este caso nuestro destino es Google Cloud Storage y BigQuery, al cual le dimos el valor 1 en nuestro archivo.\nfile_size=os.stat(props['archivo_origen']).st_size\nif (int(file_size)>0):\n   if(int(props['Destino'])=1):        \n#Tenemos las variables siguientes, sus valores son devueltos por la función upload_GCS_BQ:\n#exit_codeBQ  - Bandera para indicar si la ingesta fue exitosa o no.\n#registros    - Almacena el numero de registros del archivo.\n#Timestamp_date – La fecha en que se hace la ingesta.\n#strerror  - Si hay error en la ingesta, esta variable almacena el #código del error\n                           exit_codeBQ,registros,Timestamp_Date,strerror=upload_GCS_BQ(creds,props,item['id'])\nelse:\n    print('archivo vacio')\n#Al final del proceso, eliminamos los archivos descargados a nuestro servidor, para liberar el espacio ocupado\nfile_name=str(props.get('archivo_origen')).split('.')\n    fname = file_name[0]+'.*'\n    r = glob.glob(fname) #función usada por python para buscar archivos\n    for i in r:\n        print('Eliminando..'+str(i))\n        os.remove(i)\n <\/code><\/pre>\n
A continuaci\u00f3n, el c\u00f3digo de la funci\u00f3n upload_GCS_BQ <\/strong>el cual realiza la ingesta del archivo al proyecto de Google Cloud definido en el archivo de configuraci\u00f3n.<\/p>\n
#Librerias de GCP \nfrom google.cloud import bigquery\nfrom google.cloud import storage\nfrom google.api_core.exceptions import BadRequest\nfrom google.cloud.exceptions import NotFound\nfrom apiclient.errors import HttpError\n#Biblioteca de Python para manejo de archivos csv\nimport csv\ndef upload_GCS_BQ(creds,props,file_id):\n    exit_codeBQ=0\n    strerror=""\n    registros=0\n    Timestamp_Date = datetime.datetime.today().strftime('%Y-%m-%d %H:%M:%S.%f %Z') # obtenemos la fecha de sistema en formato Timestamp\n        #Se realiza la carga a Google Cloud Storage\n        Current_Date = datetime.datetime.today().strftime ('%Y-%b-%d %H_%M_%S')\n        #Dentro de props, vienen las propiedades del archivo\n        #a cargar, dividimos el nombre del archivo para agregarle \n        #la fecha y así crear un archivo de respaldo\n        if props.get('archivo_origen').find('.')!=-1:\n            file_part=props.get('archivo_origen').split('.',1)\n            filename_bkp=file_part[0]+' '+str(Current_Date)+'.'+file_part[1]\n        else:\n            filename_bkp=props.get('archivo_origen')+str(Current_Date)\n        #usando funciones de las bibliotecas de google se realiza la carga del archivo a Google Cloud Storage\n        try:\n            bucket = creds.get('clientGS').get_bucket(props.get('Bucket_GCS'))    \n            blob = bucket.blob(props.get('Path_GCS')+props.get('archivo_origen'))\n            blob.upload_from_filename(props.get('archivo_origen'))\n            registros=0\n            dest_bucket = creds.get('clientGS').get_bucket(props.get('Bucket_GCS'))\n            new_blob_name=props.get('Path_GCS_bkp')+filename_bkp\n            new_blob = bucket.copy_blob(\n                             blob, dest_bucket, new_blob_name)\n            #Seteamos la variable exit_codeBQ en 1 para validar que la carga fue exitosa\n            exit_codeBQ=1\n        #si hay errores en la carga se setea la variable a 0\n        except BadRequest as e:\n            for err in e.errors:\n                error=err\n            exit_codeBQ=0\n <\/code><\/pre>\n
La segunda parte de la funci\u00f3n realiza la carga a BigQuery, a partir del archivo que ya est\u00e1 en nuestro bucket de Google Cloud Storage<\/p>\n
# Configuramos las opciones de la tabla definidas en el API de BigQuery\n        dataset_ref =   creds.get('clientBQ').dataset(str(props.get('DataSet_BQ')))\n        job_config = bigquery.LoadJobConfig()\n        job_confighis = bigquery.LoadJobConfig()\n        job_config.skip_leading_rows = 1\n        job_confighis.skip_leading_rows=1\n        job_config.field_delimiter = '|'\n        job_confighis.field_delimiter = '|'\n        job_config.write_disposition = 'WRITE_TRUNCATE'\n        job_confighis.write_disposition = 'WRITE_APPEND'\n        job_config.autodetect=True\n        job_confighis.autodetect=True\n#Establecemos el formato de origen de nuestro archivo como CSV\n        job_config.source_format = bigquery.SourceFormat.CSV\n        job_confighis.source_format = bigquery.SourceFormat.CSV\n        uri = "gs:\/\/"+props.get('Bucket_GCS')+"\/"+props.get('Path_GCS')+props.get('archivo_origen') #Este es el path de nuestro archive en Cloud Storage\n        try:\n            load_job = creds.get('clientBQ').load_table_from_uri(\n                uri, dataset_ref.table(props.get('Tabla')), job_config=job_config)  # API request\n            load_job.result()  #Espera a que termine la carga de la tabla.\n            destination_table = creds.get('clientBQ').get_table(dataset_ref.table(props.get('Tabla')))\n            registros=destination_table.num_rows\n#Obtenemos el id de la tabla a partir de las propiedades definidas            \ntable_id=str(props.get('proyecto')) +'.'+str(props.get('DataSet_BQ'))+'.'+str(props.get('Tabla'))\n            table = creds.get('clientBQ').get_table(table_id)  \n#Agregamos un campo para colocar la fecha de modificación de la tabla\n            original_schema = table.schema\n            new_schema = original_schema[:]  # Creates a copy of the schema.\n            new_schema.append(bigquery.SchemaField("FECHA_MODIFICACION", "TIMESTAMP"))\n            table.schema = new_schema\n            table = creds.get('clientBQ').update_table(table, ["schema"])  \n#Hacemos un update para agregar la fecha de modificación\nqueryUpdate="UPDATE "+str(props.get('DataSet_BQ'))+"."+str(props.get('Tabla')) +" SET FECHA_MODIFICACION = TIMESTAMP('"+Timestamp_Date.strip() +"') WHERE TRUE"\n            dml_statement = ("UPDATE "+str(props.get('DataSet_BQ'))+"."+str(props.get('Tabla')) +" SET FECHA_MODIFICACION = TIMESTAMP('"+Timestamp_Date.strip() +"') WHERE TRUE")\n            query_job = creds.get('clientBQ').query(dml_statement)  \n            query_job.result()  \n                     #Seteamos la variable exit_codeBQ en 1 para validar que la carga fue exitosa\n            exit_codeBQ=1\n  #si hay errores en la carga se setea la variable a 0\nexcept BadRequest as e:\n            for err in e.errors:\n                strerror=str(err)\n            exit_codeBQ=0\n    #Con este return devolvemos los valores de cada variable a la función principal\n    return exit_codeBQ,registros,Timestamp_Date,strerror\n <\/code><\/pre>\n
Este es el C\u00f3digo b\u00e1sico para cargar nuestros archivos en Google Cloud Storage y Big Query, haciendo uso de las funciones incluidas en sus APIs.<\/p>\n
Para mayor referencia de su uso, puedes consultar los siguientes enlaces: <\/strong><\/p>\n
Google Drive: https:\/\/developers.google.com\/drive\/api\/v2\/about-sdk<\/a>Google Cloud Storage: https:\/\/cloud.google.com\/storage\/docs\/reference\/libraries#client-libraries-usage-python<\/a>Google Big Query: https:\/\/cloud.google.com\/bigquery\/docs\/reference\/libraries#client-libraries-usage-python<\/a>Cargar un archivo CSV desde Cloud Storage:https:\/\/cloud.google.com\/bigquery\/docs\/loading-data-cloud-storage-csv<\/a><\/p>\n
\u00bfQuieres saber m\u00e1s de lo que ofrecemos y ver otros casos de \u00e9xito?<\/h5>\n
\t\t\t
\n\t\t\t\t\t\tDESCUBRE BLUETAB
\n\t\t\t\t\t<\/a>
\n\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\tShare on twitter
\n\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\tShare on linkedin<\/p>\n
SOLUCIONES, <\/b>SOMOS EXPERTOS<\/p>\n
\t\t\t\t\t<\/p>\n
DATA STRATEGY<\/h5>\n
\t\t\t\t\t\t<\/a>
\n\t\t\t\t\t<\/p>\n
DATA FABRIC<\/h5>\n
\t\t\t\t\t\t<\/a>
\n\t\t\t\t\t<\/p>\n
AUGMENTED ANALYTICS<\/h5>\n
\t\t\t\t\t\t<\/a><\/p>\n
Te puede interesar<\/p>\n","protected":false},"excerpt":{"rendered":"
Mi experiencia en el mundo de Big Data – Parte II David Emmanuel Reyes N\u00fa\u00f1ez Senior Data Engineer Share on twitter Share on linkedin En la entrega anterior (adjunto) creamos los scripts para enlistar y descargar archivos desde Google Drive hacia nuestro filesystem local. En esta entrega continuaremos con el c\u00f3digo de la funci\u00f3n processDriveFiles.py […]<\/p>\n","protected":false},"author":2,"featured_media":20800,"comment_status":"open","ping_status":"open","sticky":false,"template":"elementor_header_footer","format":"standard","meta":{"_acf_changed":false,"footnotes":""},"categories":[7,30],"tags":[],"class_list":["post-13138","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-blog-es","category-tech-en"],"acf":[],"jetpack_featured_media_url":"https:\/\/bluetab.mx\/wp-content\/uploads\/2021\/10\/emanuel-scaled.png","_links":{"self":[{"href":"https:\/\/bluetab.mx\/en\/wp-json\/wp\/v2\/posts\/13138","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/bluetab.mx\/en\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/bluetab.mx\/en\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/bluetab.mx\/en\/wp-json\/wp\/v2\/users\/2"}],"replies":[{"embeddable":true,"href":"https:\/\/bluetab.mx\/en\/wp-json\/wp\/v2\/comments?post=13138"}],"version-history":[{"count":0,"href":"https:\/\/bluetab.mx\/en\/wp-json\/wp\/v2\/posts\/13138\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/bluetab.mx\/en\/wp-json\/wp\/v2\/media\/20800"}],"wp:attachment":[{"href":"https:\/\/bluetab.mx\/en\/wp-json\/wp\/v2\/media?parent=13138"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/bluetab.mx\/en\/wp-json\/wp\/v2\/categories?post=13138"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/bluetab.mx\/en\/wp-json\/wp\/v2\/tags?post=13138"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}