Skip to content

B
bcom-tp-etl-transformation-pipelines
Commit c4edffad authored by Cristian Aguirre's avatar Cristian Aguirre
Browse files
Options

Fix bugs

parent 3170e104
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 441 additions and 78 deletions
dags/components/Extractor.py
View file @ c4edffad
...@@ -163,13 +163,13 @@ def extract_from_source(command, source_conn, intern_conn, chunksize: int, timez ...@@ -163,13 +163,13 @@ def extract_from_source(command, source_conn, intern_conn, chunksize: int, timez
data.append(row) data.append(row)
if len(data) == chunksize: if len(data) == chunksize:
dataframe = pd.DataFrame(data, columns=columns_name) dataframe = pd.DataFrame(data, columns=columns_name)
save = save_from_dataframe(dataframe, tablename, intern_conn.engine) save = save_from_dataframe(dataframe, tablename, intern_conn)
if save: if save:
logger.debug(f"Guardado correctamente dataframe. Procesando más bloques") logger.debug(f"Guardado correctamente dataframe. Procesando más bloques")
data.clear() data.clear()
if len(data) > 0: if len(data) > 0:
dataframe = pd.DataFrame(data, columns=columns_name) dataframe = pd.DataFrame(data, columns=columns_name)
save = save_from_dataframe(dataframe, tablename, intern_conn.engine) save = save_from_dataframe(dataframe, tablename, intern_conn)
if save: if save:
logger.debug(f"Migrado correctamente todos los datos") logger.debug(f"Migrado correctamente todos los datos")
extract_tables.append(tablename) extract_tables.append(tablename)
...@@ -183,13 +183,13 @@ def extract_from_source(command, source_conn, intern_conn, chunksize: int, timez ...@@ -183,13 +183,13 @@ def extract_from_source(command, source_conn, intern_conn, chunksize: int, timez
data.append(row) data.append(row)
if len(data) == chunksize: if len(data) == chunksize:
dataframe = pd.DataFrame(data, columns=columns_name) dataframe = pd.DataFrame(data, columns=columns_name)
save = save_from_dataframe(dataframe, tablename, intern_conn.engine) save = save_from_dataframe(dataframe, tablename, intern_conn)
if save: if save:
logger.debug(f"Guardado correctamente dataframe. Procesando más bloques") logger.debug(f"Guardado correctamente dataframe. Procesando más bloques")
data.clear() data.clear()
if len(data) > 0: if len(data) > 0:
dataframe = pd.DataFrame(data, columns=columns_name) dataframe = pd.DataFrame(data, columns=columns_name)
save = save_from_dataframe(dataframe, tablename, intern_conn.engine) save = save_from_dataframe(dataframe, tablename, intern_conn)
if save: if save:
logger.debug(f"Migrado correctamente todos los datos") logger.debug(f"Migrado correctamente todos los datos")
......
dags/procedure_prueba.json
View file @ c4edffad
This diff is collapsed.
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment