Forum Discussion
Skilled SharerUse Dataflow Gen2 for multiple tables ingest dinamically
Memorable Memberamaaiia How about this approach:
CREATE OR ALTER VIEW vTABLE_NAMES
AS
SELECT TABLE_NAME
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = 'dbo'
;
If you run this SQL script from your SQL analytics endpoint, it will create a view in your LH. Then from you DF, all you have to do is create a query that uses this view as its source. The view return a column of all the tables in the dbo schema of the LH as a table. Now your M code can reference this table and Bob's your uncle.
Skilled SharerHi Element115
So imagine my view has 10 rows (10 tables to ingest from my SQL Server). Once my query (let's say Q1) gets the 10 rows, how can I tell DF to create a new query from each row of the result of Q1? So each new query (QN) gets data from my SQL Server and saves it in a Lakehouse table in append mode?
- Element115
amaaiia Before I answer, let me clarify something...
1__do you want to perform an incremental refresh on the tables?
2__are these tables always going to be the same? in other words, once you have created your 10 tables, it will
alsosorry--always be these 10 tables receiving data or, in future, would you have to create 1 or more new tables?- amaaiia
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.
- Element115Ok 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 pipelineOr did I miss something?