Help with notebooks / multiple .csv files for complete newbie

[Scott_Powell]

Hi, I have a need to ingest a ton of Power BI audit logs into Fabric. The eventual target is a Lakehouse table or tables. I have one log file per day (as csv), about 500 altogether. These files need extensive reshaping - things like pivoting attribute:value pairs into columns, parsing the text, etc. There's also reasonably large - about 40 or 50 million total lines, and maybe 4 or 5 million distinct audit log entries (see image below of a single entry)

I was going to do this using a Gen2 dataflow, but I can't figure out how to get a Gen2 dataflow to ingest a folder of files. Works perfectly with our on-prem network file shares, but in Fabric the Gen2 dataflow seems to want only one file at a time, no way to combine things under a Lakehouse Files section.

I'm a complete newbie on notebooks and really barely understand any Python (working to rectify that, going to take some time).

Here's a sample log file entry:

Things I've tried:

1. Using some of the sample code in the "learn fabric" paths to combine the files and publish to a Delta lakehouse table. This doesn't really work - if set VORDER on it screws up the order of lines in the audit log, making it impossible to associate lines with the proper audit log ID. If I turn VORDER off, querying the resulting table returns "Failed to read parquet file because the column segment for column '_c0' is too large"

2. I've tried using code like the following to append the 500 individual csv files into a single file. But it doesn't work - it seems instead to create a folder that has a ton of separate files in it instead of a single large file

df = spark.read.format("csv").load('Files/Power BI audit logs/Audit log*.csv')

df.write.mode("overwrite").format("csv").save("Files/singleFile")

Any ideas or guidance would be greatly appreciated - thanks!

Scott

[jwinchell40]

@Scott_Powell -

I read a lot of files data that are fed into OneLake via API calls to external systems.  Even though my data source are .json files; the code below will work for .csv files as well. 

#Imports and Includes to use different functions down the line from pyspark.sql.functions import * from pyspark.sql.window import * from pyspark.sql.types import DateType from pyspark.sql.functions import to_date,to_timestamp,sequence from delta.tables import * ##Define a Parameter Cell in the Notebook to be able to dynamically pass a file struct _year = "9999" _month = "5" _day = "2" ##Generating the path to my files dynamically ##Instead of *.json you can use *.csv as this will look for any file that ends .csv for the path specific. _root = "Files/XCM/TaskSignOff" _path = "/" + _year + "/" + "/" + _month + "/" + _day + "/*.json" _full = _root + _path ##If your CSV file has a defined schema, I recommend defining the schema in your Notebook so you can control schema drift. I am calling out the expect elements/headers that will be in the files that I am reading. By setting the schema, if the vendor adds a new field; we do not necessarily bring it in right away. _schema = StructType( [ StructField("taskId",StringType()), StructField("signOffId",StringType()), StructField("comments",StringType()), StructField("requiredComments",StringType()), StructField("requiredDate",StringType()), StructField("requiredName",StringType()), StructField("signOffBy",StringType()), StructField("signOffComments",StringType()), StructField("signOffDate",StringType()), StructField("signOffRequired",StringType()), StructField("signOffName",StringType()), StructField("statusName",StringType()), StructField("updatedBy",StringType()), StructField("updatedOn",StringType()) ] ) ##Actual Dataframe ingestion ##Change format form 'json' to 'csv' ##I also append the source path & file name to my column set _df = spark.read.format('json').schema(_schema).load(_full).select("*","_metadata.file_name","_metadata.file_path") ##Display the results of the read (first 1000 rows) display(_df) ##Write to Delta Table ##Write Initial Task List To Delta _df.write.mode("append").format("delta").save("Tables/<Table_Name>")

 When you try to write to a file from spark, all of the different work threads have part of the data.  So each worker writes their piece of the pie which results in a bunch of files with guids and other oddities for names.  In order to write as a single file you would have to call the collect() statement and that would then result in all data getting combined in 1 single worker and then you could write the data.  It is a very memory intensive operation.

Hope this helps out a little bit.

[puneetvijwani]

@Scott_Powell It sounds like you have quite the challenge on your hands, Scott!

For reading in the files, Spark has ability to read multiple csv files into a single DataFrame is definitely good way to get started , However i highly suggest adding a unique ID field to each row is a smart idea to preserve order later on. ( The error you mentioned sounds more like a corrupted parquet structure error at first glance)

Since you mentioned extensive data reshaping is needed, we can work together on the transformations - whether it's parsing text fields, pivoting, or anything else. Feel free to connect and share some example rows and we can prototype the logic.

Finally i would suggest instead of output as  CSV try outputing the input csv in to Delta format in a Lakehouse architecture makes sense for your further workloads
The main thought is to maintain data integrity - both in order and values - as i suggest you move from raw CSVs to a transformed Delta table.