Analyzing Delta lake tables in Microsoft Fabric through Parquet.Net and ADLS GEN2 API
Sachin Nandanwar
This article is inspired by Sandeep Pawar’s earlier article. Sandeep in that article demonstrated an innovative approach to obtain essential delta lake metadata details across single/multiple lakehouses. This gathered data can play crucial role for analyzing the overall optimization and performance issues of a lakehouse.
The most crucial details fetched were the RowGroup, RowGroups, RowCounts across multiple RowGroups, FileSizes and the table maintenance details. His code was native to Spark and created using PySpark.
His approach made me thinking if its possible to fetch the same exact details in an external environment independent of the Spark environment. I was able to find a way majorly using a combination of Azure Storage API’s and importantly through Parequet.Net to fetch the Metadata details. In Sandeep’s approach, PyArrow was utilized to retrieve the metadata details.
A video walkthrough of the process can be found at the end of this article here.
What is Parquet.Net ?
Parquet.Net is a fully managed, safe, extremely fast .NET library to read and write Apache Parquet files designed for .NET framework. It provides a lightweight and flexible solution for handling large datasets in applications designed to run under the .NET framework. It claims to be faster than any other Python or Java library capable of handling and processing Parquet files.
It has zero dependencies and supports super fast serialization and deserialization of the native Apache Parquet format data.I have tested this aspect and indeed it is blazingly fast. In an have upcoming article I will touch base the topic.
SetUp
Install the following Nuget Packages in your C#.Net console application
dotnet add package Newtonsoft.Json --version 13.0.3
dotnet add package Microsoft.Extensions.Configuration.Json --version 9.0.1
dotnet add package Microsoft.Extensions.Configuration --version 9.0.1
dotnet add package Microsoft.Identity.Client --version 4.67.2
dotnet add package Azure.Storage.Files.DataLake --version 12.21.0
dotnet add package Spectre.Console --version 0.49.1
We will read the configuration details from appsettings.json file.
Appsettings.json
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft": "Warning",
"Microsoft.Hosting.Lifetime": "Information"
}
},
"AllowedHosts": "*",
"ClientId": "Service Principal Client Id",
"CsvLocation": "Local Location\\DeltaLakeAnalysis.csv",
"WorkSpace": "Your Workspace",
"LakeHouse" : "Your Lakehouse"
}
The app will analyze all the lakehouses across the workspace defined in the appsettigs.json file and export the details to a DeltaLakeAnalysis.csv.You can have any other name for the csv file.
If the LakeHouse setting is empty in appsettigs.jsonthen the process would analyze all the lakehouses for a given workspace.The value can be just the lakehousename or lakehousename with an extension *.LakeHouse.
The idea is here that we iterate through all the parquet files across each table of all lake houses in a given workspace. We then use Parquet.Net to fetch the metadata details of a given table through http GET method, the results of which is passed as a stream to Parquet.Net This is achieved by iterating all the parquet files of a given table and sending the stream as input to Parquet.Net.
After obtaining the table metadata details, we then iterate through the _deltalog folder to read the JSON files and collect the table maintenance information such as the total number of operations, the last datetime when these operation were executed. This is because the DESCRIBE TABLE and DESCRIBE HISTORY commands that return this data is Spark exclusive and is not available out side the Spark environment.
The advantage of this approach is that you can define and create your custom details without being dependent on DESCRIBE TABLE and DESCRIBE HISTORY commands. For example I dont think there is a way to identify the column in a table is ZOrdered by. At least I am unaware of it. If there is then I apologies for my ignorance.
Here is the sample output of the process exported to a csv file. New data is appended to the existing data in the csv file to facilitate the tracking of the earlier metrics.
Now you might ask why does those date part of the OperationsDateTime look so weird. Well, that is because the TimeStamp values in form of Ticks in the json files are not being recorded correctly.
I had highlighted the same issue in one my previous videos here. That video was related to the Azure storage API returning high level table details and the creation time it returned is also in form of Ticks and I faced the same issue.
Update : 01/22/2025 : I finally demystified the reason why the ticks were not being converted to their equivalent datetime. The reason is that the ticks are stored in Unix Time. It means the ticks are counted from 00:00:00 UTC on Thursday, 1 January 1970, which is referred to as the Unix epoch.
Instead of adding the ticks from 00:00:00 1 January 0001 we have to instead add Ticks from the Unix epoch i.e. 00:00:00 UTC on Thursday, 1 January 1970.
Now the date values are displayed correctly.
I have updated the code and added a new function to convert the ticks to datetime by taking into account the Unix epoch.
public async static Task<string> ConvertTicksToDateTime(long ticks)
{
DateTime epoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Local);
DateTime dateTime = epoch.AddMilliseconds(ticks);
return dateTime.ToString(("dd-MM-yyyy HH:mm:ss"));
}
Code
I have three classes in the project
Program.cs being the entry point into the application.
Authentication.cs returns the bearer token
HttpMethods.cs containing the various Http methods
HttpMethods class inherits Authentication class
Authentication.cs
using Microsoft.Identity.Client;
namespace Security
{
internal class Authentication
{
public static bool istokencached = false;
public static string clientId = "";
private static string[] scopes = new string[] { "https://storage.azure.com/.default" };
private static string Authority = "https://login.microsoftonline.com/organizations";
private static string RedirectURI = "http://localhost";
public async static Task<AuthenticationResult> ReturnAuthenticationResult()
{
string AccessToken;
PublicClientApplicationBuilder PublicClientAppBuilder =
PublicClientApplicationBuilder.Create(clientId)
.WithAuthority(Authority)
.WithCacheOptions(CacheOptions.EnableSharedCacheOptions)
.WithRedirectUri(RedirectURI);
IPublicClientApplication PublicClientApplication = PublicClientAppBuilder.Build();
var accounts = await PublicClientApplication.GetAccountsAsync();
AuthenticationResult result;
try
{
result = await PublicClientApplication.AcquireTokenSilent(scopes, accounts.First())
.ExecuteAsync()
.ConfigureAwait(false);
}
catch
{
result = await PublicClientApplication.AcquireTokenInteractive(scopes)
.ExecuteAsync()
.ConfigureAwait(false);
}
istokencached = true;
return result;
}
}
}
HttpMethods.cs
using System.Net.Http.Headers;
using Microsoft.Identity.Client;
namespace Http
{
internal class HttpMethods : Security.Authentication
{
public static string access_token = "";
public static double currentfilelength=0;
public static readonly HttpClient client = new HttpClient();
protected HttpClient Client => client;
public async static Task<string> GetAsync(string url)
{
AuthenticationResult result = await ReturnAuthenticationResult();
client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", result.AccessToken);
access_token = result.AccessToken;
HttpResponseMessage response = await client.GetAsync(url);
try
{
response.EnsureSuccessStatusCode();
return await response.Content.ReadAsStringAsync();
}
catch
{
Console.WriteLine(response.Content.ReadAsStringAsync().Result);
return null;
}
}
public async static Task<Stream> SendAsync(HttpRequestMessage httprequestMessage)
{
AuthenticationResult result = await ReturnAuthenticationResult();
client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", result.AccessToken);
HttpResponseMessage response = await client.SendAsync(httprequestMessage);
response.EnsureSuccessStatusCode();
try
{
currentfilelength = (double)response.Content.Headers.ContentLength;
return await response.Content.ReadAsStreamAsync();
}
catch
{
Console.WriteLine(response.Content.ReadAsByteArrayAsync().Result);
return null;
}
}
}
}
Program.cs
using Microsoft.Extensions.Configuration;
using Newtonsoft.Json.Linq;
using Parquet;
using Parquet.Schema;
using Spectre.Console;
using File = System.IO.File;
namespace AnaylzeDeltaTables
{
internal class Program
{
private static string RedirectURI = "http://localhost";
private static string workSpace = "";//"Medallion Architecture
private static string lakeHouse = "";
public static double filelength;
public static string FolderName;
public static int NoOfFiles = 0;
public static long NoOfRows = 0;
public static long NoOfColumns = 0;
public static int RowGroupCount = 0;
public static double TotalRowGroupSize = 0;
public static string FileName;
public static string zOrderBy;
public static string partitionby;
private static int writeoperation;
private static string lastwriteoperationdatetime;
private static int optimizeoperation;
private static string lastoptimizeoperationdatetime;
private static string lastvacuumoperationdatetime;
private static int mergeoperation;
private static int vacummoperation;
private static int count;
private static string CsvLocation = "";
private static double minRowGroupSize;
private static double maxRowGroupSize;
private static double avgRowGroupSize;
public static string IsVOrderEnabled = "false";
private static int deleteoperation;
private static List<long> RowGroupSizelst = new List<long>();
private static string dfsendpoint = "";
private static string responsename = "";
private static string[] scopes = new string[] { "https://storage.azure.com/.default" };
private static string Authority = "https://login.microsoftonline.com/organizations";
static async Task Main(string[] args)
{
Http.HttpMethods.client.Timeout = TimeSpan.FromMilliseconds(-1);
ReadConfig();
string foldername = "";
if (lakeHouse != "")
{
if (!lakeHouse.Contains(".LakeHouse"))
{ lakeHouse = lakeHouse + ".LakeHouse"; }
dfsendpoint = $"https://onelake.dfs.fabric.microsoft.com/{workSpace}/{lakeHouse}/Tables?resource=filesystem&recursive=false";
await TraverseAllDirectoriesInLakeHouse(dfsendpoint);
}
else
{
JObject jsonObject_0 = await TraverseAllLakeHousesInWorkspace(workSpace);
JArray pathsArray_0 = (JArray)jsonObject_0["paths"];
foreach (JObject path_0 in pathsArray_0)
{
if (path_0["name"].ToString().Contains(".Lakehouse"))
{
lakeHouse = path_0["name"].ToString();
dfsendpoint = $"https://onelake.dfs.fabric.microsoft.com/{workSpace}/{lakeHouse}/Tables?resource=filesystem&recursive=false";
await TraverseAllDirectoriesInLakeHouse(dfsendpoint);
}
}
AnsiConsole.MarkupLine("");
AnsiConsole.MarkupLine($"[Blue]Process completed successfully..[/]");
AnsiConsole.MarkupLine("");
AnsiConsole.MarkupLine($"Csv generated at location [Red]{CsvLocation}[/]");
Thread.Sleep(2000);
}
}
public static void ReadConfig()
{
var builder = new ConfigurationBuilder()
.AddJsonFile($"appsettings.json", true, true);
var config = builder.Build();
Security.Authentication.clientId = config["ClientId"];
workSpace = config["WorkSpace"];
lakeHouse = config["LakeHouse"];
CsvLocation = config["CsvLocation"];
}
public static async Task<List<string>> TraverseAllDirectoriesInLakeHouse(string dfsendpoint)
{
string response = await Http.HttpMethods.GetAsync(dfsendpoint);
JObject jsonObject_dir = JObject.Parse(response);
JArray dirArray = (JArray)jsonObject_dir["paths"];
List<string> lst = new List<string>();
foreach (JObject dir in dirArray)
{
if (dir["isDirectory"] != null && !dir["name"].ToString().Contains("_delta_log"))
{
await TraverseAllFilesInDirectory(await GetFileName(dir["name"].ToString()));
}
}
return lst;
}
public static async Task<JObject> TraverseAllLakeHousesInWorkspace(string path)
{
string responsename = "";
string dfsendpoint = $"https://onelake.dfs.fabric.microsoft.com/{path}" + $"?resource=filesystem&recursive=false";
responsename = await Http.HttpMethods.GetAsync(dfsendpoint);
JObject jsonObject_lakhouse = JObject.Parse(responsename);
return jsonObject_lakhouse;
}
public static async Task<string> TraverseAllFilesInDirectory(string table)
{
string responsename;
dfsendpoint = $"https://onelake.dfs.fabric.microsoft.com/{workSpace}/{lakeHouse}/Tables/{table}?resource=filesystem&recursive=true";
responsename = await Http.HttpMethods.GetAsync(dfsendpoint);
string parentfolder = "";
string foldername_n = "";
if (responsename != "")
{
JObject jsonObject = JObject.Parse(responsename);
JArray pathsArray = (JArray)jsonObject["paths"];
AnsiConsole.MarkupLine("");
AnsiConsole.MarkupLine($"Process started for table [Red]{table}...[/]Please wait.");
Thread.Sleep(1000);
AnsiConsole.MarkupLine("");
foreach (JObject path_ in pathsArray)
{
if (path_["name"].ToString().Contains("parquet") && path_["IsDirectory"] == null)
{
FolderName = await GetFolderName(path_["name"].ToString());
FileName = await GetFileName(path_["name"].ToString());
string message = "";
if (FolderName.Contains("/"))
{ message = $"Processing of file [Red]{FileName}[/] started for partition [Red]{FolderName}[/]"; }
else
{ message = $"Processing of file [Red]{FileName}[/] started for table [Red]{FolderName}[/]"; }
await ReadFile($"https://onelake.dfs.fabric.microsoft.com/{workSpace}/{lakeHouse}/Tables/{FolderName}/{FileName}");
AnsiConsole.MarkupLine($"Process of parquet file [Yellow]{FileName}[/] completed for [Red]{FolderName}[/]");
NoOfFiles++;
}
if (path_["name"].ToString().Contains("json") && path_["IsDirectory"] == null)
{
FileName = await GetFileNameForJson(path_["name"].ToString());
await ReadDeltaLog($"https://onelake.dfs.fabric.microsoft.com/{workSpace}/{lakeHouse}/Tables/{table}/_delta_log/{FileName}");
}
}
WriteToCsv(CsvLocation, "LakeHouse||" + lakeHouse + "~Table||" + table + "~NoOfColumns||" + NoOfColumns + "~VOrderEnabled||" + IsVOrderEnabled + "~NoOFiles||" + NoOfFiles + "~TotalSizeOfFiles(MB)||" + filelength + "~RowGroupCount||" + RowGroupCount + "~RowGroupSize(MB)||" + TotalRowGroupSize + "~MinRowGroupSize(MB)||" + minRowGroupSize + "~MaxRowGroupSize(MB)||" + maxRowGroupSize + "~AvgRowGroupSize(MB)||" + avgRowGroupSize + "~NoOfRows||" + NoOfRows + "~partitionBy||" + partitionby + "~zOrderBy||" + zOrderBy + "~NoOfOptmizeOperations||" + optimizeoperation + "~LastOptmizeOperationDateTime||" + lastoptimizeoperationdatetime + "~NoOfWriteOperations||" + writeoperation + "~LastWriteOperationDateTime||" + lastwriteoperationdatetime + "~NoOfMergeOperations||" + mergeoperation + "~NoOfDeleteOperations||" + deleteoperation + "~NoOfVacummeOperations||" + vacummoperation + "~LastVacuumOperationDateTime||" + lastvacuumoperationdatetime);
ResetAllValues();
return foldername_n;
}
else
{
return null;
}
}
public static void ResetAllValues()
{
RowGroupCount = 0;
NoOfFiles = 0;
NoOfRows = 0;
NoOfColumns = 0;
IsVOrderEnabled = "false";
zOrderBy = "N/A";
partitionby = "N/A";
optimizeoperation = 0;
lastoptimizeoperationdatetime = "";
vacummoperation = 0;
writeoperation = 0;
lastwriteoperationdatetime = "";
lastvacuumoperationdatetime = "";
mergeoperation = 0;
deleteoperation = 0;
TotalRowGroupSize = 0;
minRowGroupSize = 0;
maxRowGroupSize = 0;
avgRowGroupSize = 0;
count = 0;
filelength = 0;
Http.HttpMethods.currentfilelength = 0;
RowGroupSizelst.Clear();
}
public static async Task ReadFile(string filePath)
{
var downloadMessage = new HttpRequestMessage
{
Method = HttpMethod.Get,
RequestUri = new Uri(filePath)
};
var response = await Http.HttpMethods.SendAsync(downloadMessage);
ParquetReader parquetReader = await ParquetReader.CreateAsync(response);
ParquetSchema schema = parquetReader.Schema;
int i = 0;
foreach (IParquetRowGroupReader rd in parquetReader.RowGroups)
{
long currentsize = parquetReader.RowGroups[i].RowGroup.TotalByteSize;
RowGroupSizelst.Add(currentsize);
count = count + 1;
i++;
};
NoOfRows += parquetReader.Metadata.NumRows;
NoOfColumns = schema.DataFields.Count();
RowGroupCount += parquetReader.RowGroupCount;
TotalRowGroupSize = (double)(RowGroupSizelst.Sum()) / 1000000;
minRowGroupSize = (double)(RowGroupSizelst.Min()) / 1000000;
maxRowGroupSize = (double)(RowGroupSizelst.Max()) / 1000000;
avgRowGroupSize = (double)((TotalRowGroupSize / count));
filelength += (double)Http.HttpMethods.currentfilelength / 1000000;
try
{
IsVOrderEnabled = parquetReader.CustomMetadata["com.microsoft.parquet.vorder.enabled"];
}
catch { IsVOrderEnabled = "false"; };
}
public static async Task ReadDeltaLog(string filePath)
{
var response = await Http.HttpMethods.GetAsync(filePath);
string[] jsonLines = response.Split(new[] { '\n', '\r' }, StringSplitOptions.RemoveEmptyEntries);
foreach (string line in jsonLines)
{
try
{
JObject commitInfo = null;
JObject jdeltalog = JObject.Parse(line);
if (jdeltalog["commitInfo"] != null)
{
commitInfo = (JObject)jdeltalog["commitInfo"];
if (commitInfo["operation"]?.ToString() == "WRITE")
{ writeoperation++; }
if (commitInfo["operation"]?.ToString() == "OPTIMIZE")
{ optimizeoperation++; }
if (commitInfo["operation"]?.ToString() == "MERGE")
{ mergeoperation++; }
if (commitInfo["operation"]?.ToString() == "DELETE")
{ deleteoperation++; }
if (commitInfo["operation"]?.ToString() == "VACUUM END")
{ vacummoperation++; }
if (commitInfo["timestamp"].ToString() != null && commitInfo["operation"]?.ToString() == "WRITE")
{
lastwriteoperationdatetime = await ConvertTicksToDateTime((long)commitInfo["timestamp"]);
}
if (commitInfo["timestamp"].ToString() != null && commitInfo["operation"]?.ToString() == "OPTIMIZE")
{
lastoptimizeoperationdatetime = await ConvertTicksToDateTime((long)commitInfo["timestamp"]);
}
if (commitInfo["timestamp"].ToString() != null && commitInfo["operation"]?.ToString() == "VACUUM END")
{
lastvacuumoperationdatetime = await ConvertTicksToDateTime((long)commitInfo["timestamp"]);
}
if (commitInfo["operationParameters"] != null)
{
JObject operationParameters = (JObject)commitInfo["operationParameters"];
if (operationParameters["zOrderBy"] != null)
{ zOrderBy = operationParameters["zOrderBy"].ToString(); }
if (operationParameters["partitionBy"] != null)
{ partitionby = operationParameters["partitionBy"].ToString(); }
}
}
}
catch (Exception ex)
{
AnsiConsole.MarkupLine($"Process errored with exception [Red]{ex.Message}...[/].");
Thread.Sleep(1000);
}
}
}
public static async void WriteToCsv(string path, string values)
{
string delimiter = ", ";
string[] parts = values.Split('~');
if (!File.Exists(path))
{
string createText = "LakeHouse " + delimiter + "Table " + delimiter + "NoOfColumns" + delimiter + "VOrderEnabled" + delimiter + "NoOfFiles" + delimiter + "TotalSizeOfFiles(MB)" + delimiter + "RowGroupCount" + delimiter + "TotalRowGroupSize(MB)" + delimiter + "MinRowGroupSize(MB)" + delimiter + "MaxRowGroupSize(MB)" + delimiter + "AvgRowGroupSize(MB)" + delimiter + "NoOfRows" + delimiter + "partitionBy" + delimiter + "zOrderBy" + delimiter + "NoOfOptmizeOperations" + delimiter + "LastOptmizeOperationDateTime" + delimiter + "NoOfWriteOperations" + delimiter + "LastWriteOperationsDateTime" + delimiter + "NoOfMergeOperations" + delimiter + "NoOfDeleteOperations" + delimiter + "NoOfVacuumOperations" + delimiter + "LastVacuumOperationsDateTime" + delimiter + Environment.NewLine;
File.WriteAllText(path, createText);
}
string appendText = parts[0].Split("||")[1].ToString() /*LakeHouse*/ + delimiter + await ReplaceInvalidValues(parts[1].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[2].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[3].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[4].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[5].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[6].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[7].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[8].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[9].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[10].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[11].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[12].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[13].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[14].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[15].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[16].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[17].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[18].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[19].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[20].Split("||")[1].ToString()) + delimiter + await ReplaceInvalidValues(parts[21].Split("||")[1].ToString()) + delimiter + Environment.NewLine;
File.AppendAllText(path, appendText);
}
public async static Task<string> ConvertTicksToDateTime(long ticks)
{
DateTime epoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Local);
DateTime dateTime = epoch.AddMilliseconds(ticks);
return dateTime.ToString(("dd-MM-yyyy HH:mm:ss"));
}
public static async Task<string> GetFileName(string value)
{
int lastSlashIndex_n = value.ToString().LastIndexOf('/');
string dir_n = value.ToString().Substring(lastSlashIndex_n + 1);
return dir_n;
}
public static async Task<string> ReplaceInvalidValues(string s)
{
if (s == ("") || s == ("[]"))
return "NA";
else
return s;
}
public static async Task<string> GetFolderName(string value)
{
int startIndex = value.IndexOf("Tables/") + "Tables/".Length;
int endIndex = value.IndexOf("/part");
string dir_n = value.Substring(startIndex, endIndex - startIndex);
return dir_n;
}
public static async Task<string> GetFileNameForJson(string value)
{
string jsonValue = value.Split('/')[^1]; // Get the last part
return jsonValue;
}
}
}
Walkthrough
Conclusion
Through this article I tried demonstrating how useful and versatile parquet.net is and combined with ADLS GEN2 API the possibilities of getting almost details of your delta tables are limitless. This powerful combination helps you identify performance pit falls across the tables of your lake houses.
Thanks for reading !!!
Subscribe to my newsletter
Read articles from Sachin Nandanwar directly inside your inbox. Subscribe to the newsletter, and don't miss out.
Written by