Paste code

esquema_vpc = &#039;d_mdl_vpc_disc&#039;

def only_del(table, path):
    # seteado de variables
    table_ = table + &#039;_SG&#039;
    path_ = path.replace(table, table_)
    path_del = path_.split(bucket)[-1][1:]
    
    # se elimina la ubicacion fisica
    bucket_s3.objects.filter(Prefix=path_del).delete()
    print(table_, path_, path_del)
    
    # se elimina la tabla en athena
    #del_ = wr.catalog.delete_table_if_exists(database=esquema_vpc, table=table)
    del_ = wr.catalog.delete_table_if_exists(database=esquema_vpc, table=table_)
    

def generate_table(esquema_vpc=esquema_vpc, query_=&#039;&#039;, table=&#039;&#039;, path=&#039;&#039;, llave_=&#039;periodo&#039;, 
                   formato=&#039;Parquet&#039;, compresion=&#039;SNAPPY&#039;, grupo=&#039;athenav2&#039;):
    # seteado de variables
    table_ = table + &#039;_SG&#039;
    path_ = path.replace(table, table_)
    path_del = path_.split(bucket)[-1][1:]
    print("tabla: ", table_)
    print("uri: ", path_)
    print("ruta: ", path_del)
    # se elimina la ubicacion fisica
    bucket_s3.objects.filter(Prefix=path_del).delete()
    print(table_, path_, path_del)
    
    # se elimina la tabla en athena
    #del_ = wr.catalog.delete_table_if_exists(database=esquema_vpc, table=table)
    del_ = wr.catalog.delete_table_if_exists(database=esquema_vpc, table=table_)
    
    time.sleep(5)
    # se crea nuevamente la tabla
    result_ = wr.athena.create_ctas_table(
        sql=query_,
        database=esquema_vpc,
        ctas_table=table_,
        wait=True,
        s3_output=path_,
        storage_format=formato,
        write_compression=compresion,
        partitioning_info=[llave_],
        athena_query_wait_polling_delay=0.5,
        boto3_session=boto3.Session(),
        workgroup=grupo
    )
    return del_, result_


def generate_table_new(query_=&#039;&#039;, table=&#039;&#039;, path=&#039;&#039;, llave_=&#039;periodo&#039;, esquema_vpc=esquema_vpc, 
                   formato=&#039;Parquet&#039;, compresion=&#039;SNAPPY&#039;, grupo=&#039;athenav2&#039;):
    # seteado de variables
    table_ = table + &#039;_SG&#039;
    path_ = path.replace(table, table_)
    path_del = path_.split(bucket)[-1][1:]
    
    # se elimina la ubicacion fisica
    bucket_s3.objects.filter(Prefix=path_del).delete()
    print("DEL S3: ", table_, path_, path_del)
    
    # se elimina la tabla en athena
    #del_ = wr.catalog.delete_table_if_exists(database=esquema_vpc, table=table)
    del_ = wr.catalog.delete_table_if_exists(database=esquema_vpc, table=table_)
    print("DEL ATHENA")
    
    time.sleep(5)
    # se crea nuevamente la tabla
    
    qyery_create = """
            CREATE TABLE {}.{}
            WITH ( format = &#039;{}&#039;, 
                     parquet_compression = &#039;{}&#039;, 
                     partitioned_by = ARRAY[&#039;{}&#039;],
                     external_location= &#039;{}&#039;
                   )
            AS (
                {}
            )
    """.format(esquema_vpc, table_, formato, compresion, llave_, path_, query_)
    print(qyery_create)
    
    result_ = wr.athena.read_sql_query(
        qyery_create,
        database=esquema_vpc,
        workgroup=grupo,         
        ctas_approach=False
    )
    return del_, result_


def apply_create_new(table=&#039;X&#039;, path=&#039;X&#039;, llave=&#039;X&#039;, query=&#039;X&#039;):
    result = generate_table_new(esquema_vpc, query, table, path, llave)
    return result[0], result[1]

def apply_create(table=&#039;X&#039;, path=&#039;X&#039;, llave=&#039;X&#039;, query=&#039;X&#039;, esquema_vpc=esquema_vpc):
    result = generate_table(esquema_vpc, query, table, path, llave)
    return result[0], result[1][&#039;ctas_query_metadata&#039;].raw_payload[&#039;Status&#039;]