Forum Discussion

amaaiia's avatar
amaaiia
Icon for Skilled Sharer rankSkilled Sharer
2 years ago

Use Dataflow Gen2 for multiple tables ingest dinamically

I have a Data Pipeline to ingest in my Lakehouse several tables located in an on-premise SQL Server, and the tables are going to be daily reloaded incrementally. They have a date field, and every day...
data engineering
Data Integration
power bi
amaaiia's avatar
amaaiia
Icon for Skilled Sharer rankSkilled Sharer
2 years ago

Hi Element115 

1. Yes

2. No, more tables can be added

 

Just to sum up, I have a source_tables table in a warehouse with all the necessary information about my tables, as follows:

 

Then in my Pipeline, firstly I ingest all tables in one DFg2 (have to add them manually one by one, and whenever there is a new table to ingets, I have to update de DFg2 to add it in a new query), an then I iterate over each row from source_tables to ingest data and then do some other operations (notebooks and other activities) to each table one by one. It's like I have to synchronise both DFg2 and source_tables table to have the same tables inside. What happens is that DFg2 is not inside my ForEach ctivity because I can't pass the source table name, the destination info (lakehouse, table_name, conneciton parameters...), etc by parameter, so I only have the rest of the activities inside ForEach (not a good practise I think, because DFg2 and source_tables are working separately)

I really would like to automate tables ingest, so if I added a new table on my source_tables table, the Pipeline would automatically load it, and DFg2 could check the tables list from source_tables so the queries to ingest data would be created automatically.

 

Element115's avatar
Element115
Icon for Memorable Member rankMemorable Member
2 years ago
Ok so here is how I would do it.  I would only use Python (a PySpark notebook) for the entire ETL cycle because, if I understand you correctly, you don't know in advance the total number of tables--one day, there could suddenly be one or more new tables to process in addition to the ones already in the LH, which means you would need new DFs to process these new tables because of how the DF data destination feature works.  
 
Or you could try to do it in one DF but the complexity of the M code required, needing to merge append data from different tables into one working table, that is M query, and persisting another query as a table that would keep track of the offsets of where each data chunk ends (in the working table) for each respective table, and then reading the data out of that one working table using the offsets from the second M query with another PySpark notebook that would finally append this data to its respective table already in the LH, is a headache I wouldn't wish on anyone.
 
Going back to the main issue, for each table you have, you need a corresponding DF because when you publish to a data destination, it is the tabular output of this DF that you want to persist to the LH, and since it is not possible to put 'Choose data destination' in a loop inside of a DF, you need one DF for each incoming table.  That is way easier to do with PySpark IMHO. 
 
Ok so, all ETL done with one PySpark notebook:
 
0__in pipeline, run PySpark notebook
1__if first run ever, populate LH with meta data table, else append new table meta data to meta data table
2__read LH meta data table
3__for each table name, (because of memory considerations, use a staging LH if too much data instead of loading into memory)
if not exist in LH --> load table from on-prem source
if exist in LH --> load incrementally, ie only new data into staging LH
4__transform data accordingly
5__persist data to production LH
6__do additional stuff in pipeline

 

Or did I miss something?