Databricks API JSON Result to Collection

naveed_raza · ‎08-12-25

Hi All,

Does anybody get a chance to work on Databricks API ?

Databricks API result comes in JSON and the JSON format is different compare to other API results as here in Databricks API we are getting columns in different array of object and actual data in different array of object.

What is the best way to convert this into proper data table (collection) with headers ?

Example:

{
"schema": {
   "columns": 
      [
        {
          "name": "headername",
          "type": "string"
        },
        {
          "name": "headername2",
          "type": "date"
        }
      ]
  },
 "result": 
   {
     "data_array":[
           [
             "row1header1value",
             "row1header2value"
           ],
           [
             "row2header1value",
             "row2header2value"
           ]
         ]
   }
}

#API, #databricks, #json

naveed_raza · ‎12-01-26

Thank you so much sharing this, let me try this

naveed_raza · ‎12-01-26

looks like we have to get , EPPlus.dll and ClosedXML.dll dll files

simone.tomaCNH · ‎12-01-26

They can be easily found in Visual Studio with Manage NuGet Packages. However, they are DLLs because the object generates an Excel file if the CreateExcel flag is set to true. Below is the version without those DLLs.

<type>Application</type>

<basetype>Application</basetype>

<datatype>unknown</datatype>

<diagnose>False</diagnose>

</element>

</appdef>

<view>

</view>

<name>Clean Up</name>

<view>

</view>

</subsheet>

<name>Authentication</name>

<view>

</view>

</subsheet>

<name>Execute Query</name>

<view>

</view>

</subsheet>

<name>Execute Query - Using Token</name>

<view>

</view>

</subsheet>

</stage>

</stage>

<reference>System.dll</reference>

<reference>System.Data.dll</reference>

<reference>System.Xml.dll</reference>

<reference>System.Drawing.dll</reference>

<reference>System.Net.Http.dll</reference>

<reference>netstandard.dll</reference>

<reference>System.Core.dll</reference>

<reference>System.Net.Http.Json.dll</reference>

<reference>Newtonsoft.Json.dll</reference>

<reference>System.Windows.Forms.dll</reference>

<reference>System.Net.Http.dll</reference>

</references>

<import>System</import>

<import>System.Drawing</import>

<import>System.Data</import>

<import>System.Net.Http</import>

<import>System.Text</import>

<import>System.Collections.Generic</import>

<import>System.Linq</import>

<import>System.IO</import>

<import>Newtonsoft.Json</import>

<import>Newtonsoft.Json.Linq</import>

<import>System.Windows.Forms</import>

<import>System.Net.Http.Headers</import>

</imports>

<language>csharp</language>

<code><![CDATA[

public class Column

{

public string name { get; set; }

}

public class Row

{

public List<string> values { get; set; }

}

public class JsonResponse

{

public List<Column> columns { get; set; }

public List<Row> rows { get; set; }

}

public static DataTable GenerateDataTableFromJson(string json)

{

JObject jsonObject = JObject.Parse(json);

JArray columns = (JArray)jsonObject["manifest"]["schema"]["columns"];

JArray dataArray = (JArray)jsonObject["result"]["data_array"];

DataTable dataTable = new DataTable();

foreach (var column in columns)

{

dataTable.Columns.Add(column["name"].ToString(), typeof(string));

}

foreach (var row in dataArray)

{

DataRow dataRow = dataTable.NewRow();

for (int i = 0; i < columns.Count; i++)

{

dataRow[i] = row[i].ToString();

}

dataTable.Rows.Add(dataRow);

}

return dataTable;

}

public static string EscapeForJson(string s)

{

if (s == null) return "";

return s.Replace("\\", "\\\\").Replace("\"", "\\\"");

}

]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

</inputs>

</stage>

</outputs>

</stage>

<narrative>Clean Up Page

This is an optional page where you might choose to perform some finalisation (or "cleanup") tasks as your business object is closed down.

The cleanup action will be called automatically immediately after closing your business object at the end of a business process.

You will not be able to call this action from a business process, nor will it be called at any other time than before the disposal of the business object.</narrative>

</stage>

<narrative>Initialise Page

This is an optional page where you might choose to perform some initialisation tasks after your business object is loaded.

The initialise action will be called automatically immediately after loading your business object.

You will not be able to call this action from a business process, nor will it be called at any other time than after the creation of the object.</narrative>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

JSONOutput = "";

HttpClient client = new HttpClient();

HttpRequestMessage request = new HttpRequestMessage(HttpMethod.Post, baseUrl)

{

Content = new StringContent(

$"{{\"statement\":\"{sqlQuery}\", \"warehouse_id\":\"{warehouseId}\", \"timeout_seconds\":600}}",

Encoding.UTF8,

"application/json"

)

};

request.Headers.Add("Authorization", $"Bearer {accessToken}"); // Invia la richiesta

HttpResponseMessage response = client.SendAsync(request).Result;

if (response.IsSuccessStatusCode)

{

string responseBody = response.Content.ReadAsStringAsync().Result;

Success = true;

JSONOutput = responseBody;

}

else

{

string responseBody = response.Content.ReadAsStringAsync().Result;

ErrorMessage = responseBody;

}

]]></code>

</stage>

<narrative>https://westeurope.azuredatabricks.net/api/2.0/sql/statements</narrative>

</stage>

</stage>

</stage>

</stage>

<initialvalue>False</initialvalue>

</stage>

</stage>

</stage>

<datatype>collection</datatype>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

dataTable = null;

dataTable = new DataTable();

try

{

dataTable = GenerateDataTableFromJson(json);

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = ex.Message;

}

]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</outputs>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

<datatype>password</datatype>

</initialvalueenc>

</stage>

</inputs>

</outputs>

<code><![CDATA[

string tokenUrl = $"https://login.microsoftonline.com/{TenantID}/oauth2/v2.0/token";

using (var httpClient = new HttpClient())

{

var form = new FormUrlEncodedContent(new[]

{

new KeyValuePair<string, string>("client_id", ClientId),

new KeyValuePair<string, string>("client_secret", ClientSecret),

new KeyValuePair<string, string>("grant_type", "client_credentials"),

// Scope corretto per Azure Databricks (AAD resource + /.default)

new KeyValuePair<string, string>("scope", "2ff814a6-3304-4ab8-85cb-cd0e6f879c1d/.default")

});

var resp = httpClient.PostAsync(tokenUrl, form).GetAwaiter().GetResult();

string body = resp.Content.ReadAsStringAsync().GetAwaiter().GetResult();

if (!resp.IsSuccessStatusCode)

{

throw new Exception($"Token request failed ({resp.StatusCode}): {body}");

}

var json = JObject.Parse(body);

string accessToken = json["access_token"]?.ToString();

if (string.IsNullOrEmpty(accessToken))

throw new Exception("access_token missing in token response");

AccessToken = accessToken;

}

]]></code>

</stage>

</stage>

</stage>

</inputs>

</stage>

</outputs>

</stage>

</stage>

</stage>

</stage>

<initialvalue>False</initialvalue>

</stage>

</stage>

</stage>

<datatype>collection</datatype>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

dataTable = null;

dataTable = new DataTable();

try

{

dataTable = GenerateDataTableFromJson(json);

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = ex.Message;

}

]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

<initialvalue>False</initialvalue>

</stage>

</stage>

</stage>

</stage>

</stage>

<onsuccess>0f46bfec-9fc3-4b45-9073-b2a2feebbbe3</onsuccess>

</stage>

</stage>

</stage>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

JSONOutput = "";

try

{

using (var httpClient = new HttpClient())

{

httpClient.DefaultRequestHeaders.Add("Authorization", $"Bearer {accessToken}");

var payload = new

{

statement = sqlQuery,

warehouse_id = warehouseId,

wait_timeout = "30s"

};

var content = new StringContent(Newtonsoft.Json.JsonConvert.SerializeObject(payload), Encoding.UTF8, "application/json");

var resp = httpClient.PostAsync(databricksUrl, content).GetAwaiter().GetResult();

string body = resp.Content.ReadAsStringAsync().GetAwaiter().GetResult();

if (!resp.IsSuccessStatusCode)

{

throw new Exception($"Query failed ({resp.StatusCode}): {body}");

}

else

{

JSONOutput = body;

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = ex.ToString();

}

]]></code>

</stage>

<narrative>https://adb-<workspace-id>.<shard>.azuredatabricks.net/api/2.0/sql/statements</narrative>

</stage>

</outputs>

</stage>

</process>

simone.tomaCNH · ‎20-01-26

Hi,

I have added more robust logic for the Using Token part with request verification and better management of the response JSON.

<type>Application</type>

<basetype>Application</basetype>

<datatype>unknown</datatype>

<diagnose>False</diagnose>

</element>

</appdef>

<view>

</view>

<name>Clean Up</name>

<view>

</view>

</subsheet>

<name>Authentication</name>

<view>

</view>

</subsheet>

<name>Execute Query</name>

<view>

</view>

</subsheet>

<name>Execute Query - Using Token</name>

<view>

</view>

</subsheet>

</stage>

</stage>

<reference>System.dll</reference>

<reference>System.Data.dll</reference>

<reference>System.Xml.dll</reference>

<reference>System.Drawing.dll</reference>

<reference>System.Net.Http.dll</reference>

<reference>EPPlus.dll</reference>

<reference>ClosedXML.dll</reference>

<reference>netstandard.dll</reference>

<reference>System.Core.dll</reference>

<reference>System.Net.Http.Json.dll</reference>

<reference>Newtonsoft.Json.dll</reference>

<reference>System.Windows.Forms.dll</reference>

<reference>System.Net.Http.dll</reference>

<reference>System.Threading.dll</reference>

<reference>System.Net.dll</reference>

</references>

<import>System</import>

<import>System.Drawing</import>

<import>System.Data</import>

<import>System.Net.Http</import>

<import>System.Text</import>

<import>ClosedXML.Excel</import>

<import>OfficeOpenXml</import>

<import>System.Collections.Generic</import>

<import>System.Linq</import>

<import>System.IO</import>

<import>Newtonsoft.Json</import>

<import>Newtonsoft.Json.Linq</import>

<import>System.Windows.Forms</import>

<import>System.Net.Http.Headers</import>

<import>System.Threading</import>

<import>System.Net</import>

</imports>

<language>csharp</language>

<code><![CDATA[public class DataTableToExcel

{

public void ExportToExcel(DataTable dataTable, string filePath, string sheet)

{

// Configura EPPlus per usare la licenza non commerciale

using (var package = new ExcelPackage())

{

var worksheet = package.Workbook.Worksheets.Add(sheet);

// Aggiungi le intestazioni delle colonne

for (int col = 0; col < dataTable.Columns.Count; col++)

{

worksheet.Cells[1, col + 1].Value = dataTable.Columns[col].ColumnName;

}

// Aggiungi i dati delle righe

worksheet.Cells["A2"].LoadFromDataTable(dataTable, false);

// Salva il file Excel

var fileInfo = new FileInfo(filePath);

package.SaveAs(fileInfo);

}

public class Column

{

public string name { get; set; }

}

public class Row

{

public List<string> values { get; set; }

}

public class JsonResponse

{

public List<Column> columns { get; set; }

public List<Row> rows { get; set; }

}

public static DataTable GenerateDataTableFromJson(string json)

{

var jsonObject = JObject.Parse(json);

// 1) Trova data_array (in result.data_array o alla root)

var dataArrayToken =

jsonObject.SelectToken("result.data_array") ??

jsonObject.SelectToken("data_array");

if (dataArrayToken == null || dataArrayToken.Type != JTokenType.Array)

throw new InvalidOperationException("Nessun 'data_array' trovato nel JSON.");

var dataArray = (JArray)dataArrayToken;

// 2) Prova a leggere lo schema colonne se presente

var columnsToken = jsonObject.SelectToken("manifest.schema.columns") as JArray;

var columnNames = new List<string>();

int maxCols;

if (columnsToken != null && columnsToken.Count > 0)

{

foreach (var col in columnsToken)

{

// Cerca proprietà comuni per il nome colonna

var name = (string)(col["name"] ?? col["title"] ?? col["id"]);

if (string.IsNullOrWhiteSpace(name))

name = $"Col{columnNames.Count + 1}";

columnNames.Add(name);

}

maxCols = columnNames.Count;

}

else

{

// 3) Nessuno schema: ricava il numero max di colonne dalle righe

maxCols = dataArray

.Select(r => (r as JArray)?.Count ?? 0)

.DefaultIfEmpty(0)

.Max();

for (int i = 1; i <= maxCols; i++)

columnNames.Add($"Col{i}");

}

// 4) Crea il DataTable e popola i dati

var table = new DataTable();

foreach (var name in columnNames)

table.Columns.Add(name, typeof(string));

foreach (var rowToken in dataArray)

{

var rowArray = rowToken as JArray;

if (rowArray == null)

continue; // oppure: throw, se vuoi essere più rigido

var dr = table.NewRow();

for (int i = 0; i < maxCols; i++)

{

if (i < rowArray.Count)

{

var cell = rowArray[i];

// Se vuoi stringa vuota al posto del DBNull:

// dr[i] = cell.Type == JTokenType.Null ? "" : (string)cell;

dr[i] = cell.Type == JTokenType.Null ? (object)DBNull.Value : (string)cell;

}

else

{

dr[i] = DBNull.Value;

}

table.Rows.Add(dr);

}

return table;

}

public static string EscapeForJson(string s)

{

if (s == null) return "";

return s.Replace("\\", "\\\\").Replace("\"", "\\\"");

}

public static (string StatementId, string State) ParseWithNewtonsoft(string json)

{

var jo = JObject.Parse(json);

string statementId = (string)jo["statement_id"];

string state = (string)jo["status"]?["state"];

return (statementId, state);

}

]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

</inputs>

</stage>

</outputs>

</stage>

<narrative>Clean Up Page

This is an optional page where you might choose to perform some finalisation (or "cleanup") tasks as your business object is closed down.

The cleanup action will be called automatically immediately after closing your business object at the end of a business process.

You will not be able to call this action from a business process, nor will it be called at any other time than before the disposal of the business object.</narrative>

</stage>

<narrative>Initialise Page

This is an optional page where you might choose to perform some initialisation tasks after your business object is loaded.

The initialise action will be called automatically immediately after loading your business object.

You will not be able to call this action from a business process, nor will it be called at any other time than after the creation of the object.</narrative>

</stage>

</inputs>

</outputs>

<code><![CDATA[

Success = false;

ErrorMessage = "";

JSONOutput = "";

// Funzioni locali di supporto: estrazione dettagli errore e SQLSTATE

string ExtractDetail(JObject j)

{

if (j == null) return null;

return (string)j["status"]?["error"]?["message"]

?? (string)j["error"]?["message"]

?? (string)j["status"]?["error"]?["code"]

?? (string)j["error"]?["code"];

}

string ExtractSqlState(JObject j)

{

if (j == null) return null;

return (string)j["status"]?["error"]?["sql_state"]

?? (string)j["error"]?["sql_state"];

}

void BuildShortErrorFromHttp(string forcedState, HttpResponseMessage resp, string body, out string errorMessage)

{

var stateLabel = string.IsNullOrWhiteSpace(forcedState) ? "FAILED" : forcedState;

string detail = null;

string sqlState = null;

try

{

var j = string.IsNullOrWhiteSpace(body) ? null : JObject.Parse(body);

detail = ExtractDetail(j);

sqlState = ExtractSqlState(j);

}

catch { /* ignore parse errors */ }

if (string.IsNullOrEmpty(detail))

detail = $"HTTP {(int)resp.StatusCode} {resp.ReasonPhrase}";

if (string.IsNullOrEmpty(sqlState))

sqlState = "N/A";

errorMessage = $"State: {stateLabel} - Detail: {detail} - SQLSTATE: {sqlState}";

}

HttpClient client = new HttpClient();

try

{

// ====== POST ======

HttpRequestMessage request = new HttpRequestMessage(HttpMethod.Post, baseUrl)

{

Content = new StringContent(

$"{{\"statement\":\"{sqlQuery}\", \"warehouse_id\":\"{warehouseId}\", \"timeout_seconds\":600}}",

Encoding.UTF8,

"application/json"

)

};

request.Headers.Add("Authorization", $"Bearer {accessToken}");

// Invia la richiesta

HttpResponseMessage response = client.SendAsync(request).Result;

string responseBody = response.Content.ReadAsStringAsync().Result;

if (response.IsSuccessStatusCode)

{

Success = true;

JSONOutput = responseBody;

}

else

{

BuildShortErrorFromHttp("FAILED", response, responseBody, out ErrorMessage);

Success = false;

}

/* =========================

POLLING LOGIC (ADD ONLY)

========================= */

// Procedi al polling SOLO se la POST è andata a buon fine e hai un body

if (Success && !string.IsNullOrWhiteSpace(JSONOutput))

{

JObject jo = null;

try { jo = JObject.Parse(JSONOutput); } catch { jo = null; }

if (jo == null)

{

Success = false;

ErrorMessage = "State: FAILED - Detail: Impossibile interpretare la risposta POST come JSON. - SQLSTATE: N/A";

}

else

{

string statementId = (string)jo["statement_id"];

string state = (string)jo["status"]?["state"];

// Stati terminali previsti dall’API

string[] terminal = { "SUCCEEDED", "FAILED", "CANCELED", "CLOSED" };

bool needsPolling =

!string.IsNullOrWhiteSpace(statementId) &&

(string.IsNullOrWhiteSpace(state) || Array.IndexOf(terminal, state.ToUpperInvariant()) < 0);

if (needsPolling)

{

// Timeout lato client per il ciclo di polling (regolabile)

TimeSpan pollingTimeout = TimeSpan.FromMinutes(10);

DateTime start = DateTime.UtcNow;

// Backoff esponenziale (cap a 5s)

int delayMs = 500;

// baseUrl è l’endpoint POST /sql/statements → per il GET aggiungo /{statement_id}

string getUrl = baseUrl.TrimEnd('/') + "/" + statementId;

while (true)

{

// timeout lato client

if (DateTime.UtcNow - start > pollingTimeout)

{

Success = false;

ErrorMessage = "State: FAILED - Detail: Polling scaduto lato client. - SQLSTATE: N/A";

break;

}

Thread.Sleep(delayMs);

delayMs = Math.Min(delayMs * 2, 5000); // 0.5s → 1s → 2s → 4s → 5s ...

var pollReq = new HttpRequestMessage(HttpMethod.Get, getUrl);

pollReq.Headers.Add("Authorization", $"Bearer {accessToken}");

var pollResp = client.SendAsync(pollReq).Result;

string pollBody = pollResp.Content.ReadAsStringAsync().Result;

if (pollResp.StatusCode == HttpStatusCode.NotFound)

{

Success = false;

ErrorMessage = "State: FAILED - Detail: Statement non trovato (potrebbe essere scaduto dal warehouse). - SQLSTATE: N/A";

break;

}

if (!pollResp.IsSuccessStatusCode)

{

BuildShortErrorFromHttp("FAILED", pollResp, pollBody, out ErrorMessage);

Success = false;

break;

}

// Aggiorno sempre l’ultimo body positivo

JSONOutput = pollBody;

JObject p = null;

try { p = JObject.Parse(pollBody); } catch { p = null; }

if (p == null)

{

// Continua a pollare finché non ottieni JSON valido, ma rispetta timeout

continue;

}

state = (string)p["status"]?["state"];

if (string.IsNullOrWhiteSpace(state)) continue;

string stateUpper = state.ToUpperInvariant();

// Uscita su stato terminale

if (Array.IndexOf(terminal, stateUpper) >= 0)

{

if (stateUpper == "SUCCEEDED")

{

Success = true;

ErrorMessage = "";

}

else

{

Success = false;

// Compose errore sintetico come richiesto

string detail = ExtractDetail(p) ?? $"State={state}";

string sqlState = ExtractSqlState(p) ?? "N/A";

ErrorMessage = $"State: {stateUpper} - Detail: {detail} - SQLSTATE: {sqlState}";

}

break;

}

catch (Exception ex)

{

Success = false;

// Nessuno stacktrace: solo dettaglio sintetico

ErrorMessage = $"State: FAILED - Detail: {ex.Message} - SQLSTATE: N/A";

}

finally

{

client?.Dispose();

}

]]></code>

</stage>

<initialvalue xml:space="preserve">https://westeurope.azuredatabricks.net/api/2.0/sql/statements</initialvalue>

</stage>

</stage>

</stage>

</stage>

<initialvalue>False</initialvalue>

</stage>

</stage>

</stage>

<datatype>collection</datatype>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

dataTable = null;

dataTable = new DataTable();

try

{

dataTable = GenerateDataTableFromJson(json);

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = ex.Message;

}

]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

<initialvalue>False</initialvalue>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

try

{

DataTableToExcel exporter = new DataTableToExcel();

exporter.ExportToExcel(DatatableResult, Filepath, SheetName);

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = ex.Message;

}]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</outputs>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

<datatype>password</datatype>

</initialvalueenc>

</stage>

</inputs>

</outputs>

<code><![CDATA[

string tokenUrl = $"https://login.microsoftonline.com/{TenantID}/oauth2/v2.0/token";

using (var httpClient = new HttpClient())

{

var form = new FormUrlEncodedContent(new[]

{

new KeyValuePair<string, string>("client_id", ClientId),

new KeyValuePair<string, string>("client_secret", ClientSecret),

new KeyValuePair<string, string>("grant_type", "client_credentials"),

// Scope corretto per Azure Databricks (AAD resource + /.default)

new KeyValuePair<string, string>("scope", "2ff814a6-3304-4ab8-85cb-cd0e6f879c1d/.default")

});

var resp = httpClient.PostAsync(tokenUrl, form).GetAwaiter().GetResult();

string body = resp.Content.ReadAsStringAsync().GetAwaiter().GetResult();

if (!resp.IsSuccessStatusCode)

{

throw new Exception($"Token request failed ({resp.StatusCode}): {body}");

}

var json = JObject.Parse(body);

string accessToken = json["access_token"]?.ToString();

if (string.IsNullOrEmpty(accessToken))

throw new Exception("access_token missing in token response");

AccessToken = accessToken;

}

]]></code>

</stage>

</stage>

</stage>

</inputs>

</stage>

</outputs>

</stage>

</stage>

</stage>

</stage>

<initialvalue>False</initialvalue>

</stage>

</stage>

</stage>

<datatype>collection</datatype>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

dataTable = null;

dataTable = new DataTable();

try

{

dataTable = GenerateDataTableFromJson(json);

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = ex.Message;

}

]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

<initialvalue>False</initialvalue>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

try

{

DataTableToExcel exporter = new DataTableToExcel();

exporter.ExportToExcel(DatatableResult, Filepath, SheetName);

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = ex.Message;

}]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

<onsuccess>0f46bfec-9fc3-4b45-9073-b2a2feebbbe3</onsuccess>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</inputs>

</outputs>

<code><![CDATA[

// Assumo che tu abbia già queste variabili in scope:

// string databricksUrl; // es: "https://<workspace>/api/2.0/sql/statements"

// string accessToken; // Bearer token del Service Principal

// string sqlQuery;

// string warehouseId;

Success = false;

ErrorMessage = "";

JSONOutput = "";

// == Impostazione opzionale ==

bool fetchAllChunks = true; // true = unisce tutti i chunk; false = solo chunk 0

TimeSpan overallTimeout = TimeSpan.FromMinutes(10);

int delayMs = 500;

int maxDelayMs = 5000;

// --- Helper locali identici ai tuoi ---

string ExtractDetail(JObject j)

{

if (j == null) return null;

return (string)j["status"]?["error"]?["message"]

?? (string)j["error"]?["message"]

?? (string)j["status"]?["error"]?["code"]

?? (string)j["error"]?["code"];

}

string ExtractSqlState(JObject j)

{

if (j == null) return null;

return (string)j["status"]?["error"]?["sql_state"]

?? (string)j["error"]?["sql_state"];

}

void BuildShortErrorFromHttp(string forcedState, HttpResponseMessage resp, string body, out string errorMessage)

{

var stateLabel = string.IsNullOrWhiteSpace(forcedState) ? "FAILED" : forcedState;

string detail = null;

string sqlState = null;

try

{

var j = string.IsNullOrWhiteSpace(body) ? null : JObject.Parse(body);

detail = ExtractDetail(j);

sqlState = ExtractSqlState(j);

}

catch { /* ignore parse errors */ }

if (string.IsNullOrEmpty(detail))

detail = $"HTTP {(int)resp.StatusCode} {resp.ReasonPhrase}";

if (string.IsNullOrEmpty(sqlState))

sqlState = "N/A";

errorMessage = $"State: {stateLabel} - Detail: {detail} - SQLSTATE: {sqlState}";

}

// --- Funzioni locali per i chunk ---

(bool ok, JObject resultObj, string err) FetchAndMergeChunks(

HttpClient httpClient,

string baseUrl,

string statementId,

int totalChunks,

bool fetchAll)

{

try

{

var allRows = new JArray();

int chunksToFetch = fetchAll ? totalChunks : Math.Min(1, totalChunks);

for (int i = 0; i < chunksToFetch; i++)

{

var url = $"{baseUrl.TrimEnd('/')}/{statementId}/result?chunk_index={i}";

var resp = httpClient.GetAsync(url).GetAwaiter().GetResult();

var body = resp.Content.ReadAsStringAsync().GetAwaiter().GetResult();

if (!resp.IsSuccessStatusCode)

{

BuildShortErrorFromHttp("FAILED", resp, body, out var errMsg);

return (false, null, errMsg);

}

JObject chunk = null;

try { chunk = JObject.Parse(body); } catch { }

if (chunk == null)

{

return (false, null, "State: FAILED - Detail: Impossibile interpretare la risposta dei risultati come JSON. - SQLSTATE: N/A");

}

var rows = (JArray)chunk["data_array"] ?? new JArray();

foreach (var r in rows) allRows.Add(r);

}

var resultObj = new JObject

{

["chunk_index"] = 0,

["row_offset"] = 0,

["row_count"] = allRows.Count,

["data_array"] = allRows

};

return (true, resultObj, null);

}

catch (Exception ex)

{

return (false, null, $"State: FAILED - Detail: {ex.Message} - SQLSTATE: N/A");

}

try

{

using (var httpClient = new HttpClient())

{

httpClient.DefaultRequestHeaders.Add("Authorization", $"Bearer {accessToken}");

// 1) POST: crea statement (senza attendere)

var payload = new

{

statement = sqlQuery,

warehouse_id = warehouseId,

wait_timeout = "0s",

format = "JSON_ARRAY" // così hai data_array coerente col primo metodo

};

var content = new StringContent(

JsonConvert.SerializeObject(payload),

Encoding.UTF8,

"application/json"

);

var postResp = httpClient.PostAsync(databricksUrl, content).GetAwaiter().GetResult();

var postBody = postResp.Content.ReadAsStringAsync().GetAwaiter().GetResult();

if (!postResp.IsSuccessStatusCode)

{

BuildShortErrorFromHttp("FAILED", postResp, postBody, out ErrorMessage);

Success = false;

return;

}

JObject postJson = null;

try { postJson = JObject.Parse(postBody); } catch { }

if (postJson == null)

{

ErrorMessage = "State: FAILED - Detail: Impossibile interpretare la risposta POST come JSON. - SQLSTATE: N/A";

Success = false;

return;

}

var statementId = (string)postJson["statement_id"];

var stateImmediate = (string)postJson["status"]?["state"];

if (string.IsNullOrEmpty(statementId))

{

// Caso finale immediato

if (!"SUCCEEDED".Equals(stateImmediate, StringComparison.OrdinalIgnoreCase))

{

var detail = ExtractDetail(postJson) ?? "N/A";

var sqlState = ExtractSqlState(postJson) ?? "N/A";

ErrorMessage = $"State: {stateImmediate ?? "FAILED"} - Detail: {detail} - SQLSTATE: {sqlState}";

Success = false;

return;

}

else

{

// Già completo → restituisco l'oggetto come nel primo metodo

JSONOutput = postJson.ToString(Formatting.None);

Success = true;

return;

}

// 2) Polling fino a stato terminale

var start = DateTime.UtcNow;

string finalState = null;

JObject finalStatement = null;

string[] terminal = { "SUCCEEDED", "FAILED", "CANCELED", "CLOSED" };

while (true)

{

var getUrl = $"{databricksUrl.TrimEnd('/')}/{statementId}";

var getResp = httpClient.GetAsync(getUrl).GetAwaiter().GetResult();

var getBody = getResp.Content.ReadAsStringAsync().GetAwaiter().GetResult();

if (!getResp.IsSuccessStatusCode)

{

BuildShortErrorFromHttp("FAILED", getResp, getBody, out ErrorMessage);

Success = false;

return;

}

JObject getJson = null;

try { getJson = JObject.Parse(getBody); } catch { }

if (getJson == null)

{

ErrorMessage = "State: FAILED - Detail: Impossibile interpretare la risposta GET polling come JSON. - SQLSTATE: N/A";

Success = false;

return;

}

finalStatement = getJson;

var state = (string)getJson["status"]?["state"];

if (!string.IsNullOrWhiteSpace(state) && terminal.Contains(state.ToUpperInvariant()))

{

finalState = state;

break;

}

if (DateTime.UtcNow - start > overallTimeout)

{

ErrorMessage = $"State: FAILED - Detail: Polling scaduto dopo {overallTimeout.TotalSeconds:N0} secondi. - SQLSTATE: N/A";

Success = false;

return;

}

Thread.Sleep(delayMs);

delayMs = Math.Min(delayMs * 2, maxDelayMs);

}

// 3) Esito finale

if (!"SUCCEEDED".Equals(finalState, StringComparison.OrdinalIgnoreCase))

{

var detail = ExtractDetail(finalStatement) ?? "N/A";

var sqlState = ExtractSqlState(finalStatement) ?? "N/A";

ErrorMessage = $"State: {finalState} - Detail: {detail} - SQLSTATE: {sqlState}";

Success = false;

return;

}

// 4) Recupero risultati e innesto in finalStatement

JToken resultToken = finalStatement["result"];

// Quanti chunk ci sono?

int totalChunks = 1;

try

{

totalChunks = (int?)finalStatement["manifest"]?["total_chunk_count"] ?? 1;

}

catch { /* ignore */ }

if (resultToken == null)

{

// Non c'è result nello statement → scarico i chunk e UNISCO

var (ok, resultObj, err) = FetchAndMergeChunks(httpClient, databricksUrl, statementId, totalChunks, fetchAllChunks);

if (!ok)

{

ErrorMessage = err;

Success = false;

return;

}

finalStatement["result"] = resultObj;

}

else

{

// C'è già result nello statement.

// Se vuoi essere sicuro di avere TUTTO, e ci sono più chunk, li recupero e unisco.

if (fetchAllChunks && totalChunks > 1)

{

// Riparto da zero e unisco tutti i chunk (più semplice e robusto)

var (ok, resultObj, err) = FetchAndMergeChunks(httpClient, databricksUrl, statementId, totalChunks, fetchAllChunks);

if (!ok)

{

ErrorMessage = err;

Success = false;

return;

}

finalStatement["result"] = resultObj;

}

// Allineo counters (opzionale)

try

{

var dataArray = finalStatement["result"]?["data_array"] as JArray;

if (dataArray != null)

{

finalStatement["result"]["row_count"] = dataArray.Count;

finalStatement["result"]["chunk_index"] = 0;

finalStatement["result"]["row_offset"] = 0;

}

catch { /* ignore */ }

// 5) OUTPUT: restituisco lo statement completo (come nel primo metodo)

JSONOutput = finalStatement.ToString(Formatting.None);

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = $"State: FAILED - Detail: {ex.Message} - SQLSTATE: N/A";

}

]]></code>

</stage>

<narrative>https://adb-<workspace-id>.<region-shard>.azuredatabricks.net/api/2.0/sql/statements</narrative>

</stage>

</outputs>

</stage>

</process>

naveed_raza · ‎20-01-26

thank you so much for sharing this, one more thing is , if we have large data set then API gives results in chunks,

currently i am checking total chunk count and looping based on that to fetch , store data chunk wise

simone.tomaCNH · ‎21-01-26

The chunk part was already handled in the Execute Query action with Service Principal. I also updated the execute query part with a personal token and added memory management with garbage collection.

<type>Application</type>

<basetype>Application</basetype>

<datatype>unknown</datatype>

<diagnose>False</diagnose>

</element>

</appdef>

<view>

</view>

<name>Clean Up</name>

<view>

</view>

</subsheet>

<name>Authentication</name>

<view>

</view>

</subsheet>

<name>Execute Query</name>

<view>

</view>

</subsheet>

<name>Execute Query - Using Token</name>

<view>

</view>

</subsheet>

</stage>

</stage>

<reference>System.dll</reference>

<reference>System.Data.dll</reference>

<reference>System.Xml.dll</reference>

<reference>System.Drawing.dll</reference>

<reference>System.Net.Http.dll</reference>

<reference>EPPlus.dll</reference>

<reference>ClosedXML.dll</reference>

<reference>netstandard.dll</reference>

<reference>System.Core.dll</reference>

<reference>System.Net.Http.Json.dll</reference>

<reference>Newtonsoft.Json.dll</reference>

<reference>System.Windows.Forms.dll</reference>

<reference>System.Net.Http.dll</reference>

<reference>System.Threading.dll</reference>

<reference>System.Net.dll</reference>

</references>

<import>System</import>

<import>System.Drawing</import>

<import>System.Data</import>

<import>System.Net.Http</import>

<import>System.Text</import>

<import>ClosedXML.Excel</import>

<import>OfficeOpenXml</import>

<import>System.Collections.Generic</import>

<import>System.Linq</import>

<import>System.IO</import>

<import>Newtonsoft.Json</import>

<import>Newtonsoft.Json.Linq</import>

<import>System.Windows.Forms</import>

<import>System.Net.Http.Headers</import>

<import>System.Threading</import>

<import>System.Net</import>

</imports>

<language>csharp</language>

<code><![CDATA[public class DataTableToExcel

{

public void ExportToExcel(DataTable dataTable, string filePath, string sheet)

{

// Configura EPPlus per usare la licenza non commerciale

using (var package = new ExcelPackage())

{

var worksheet = package.Workbook.Worksheets.Add(sheet);

// Aggiungi le intestazioni delle colonne

for (int col = 0; col < dataTable.Columns.Count; col++)

{

worksheet.Cells[1, col + 1].Value = dataTable.Columns[col].ColumnName;

}

// Aggiungi i dati delle righe

worksheet.Cells["A2"].LoadFromDataTable(dataTable, false);

// Salva il file Excel

var fileInfo = new FileInfo(filePath);

package.SaveAs(fileInfo);

}

public class Column

{

public string name { get; set; }

}

public class Row

{

public List<string> values { get; set; }

}

public class JsonResponse

{

public List<Column> columns { get; set; }

public List<Row> rows { get; set; }

}

public static DataTable GenerateDataTableFromJson(string json)

{

var jsonObject = JObject.Parse(json);

// 1) Trova data_array (in result.data_array o alla root)

var dataArrayToken =

jsonObject.SelectToken("result.data_array") ??

jsonObject.SelectToken("data_array");

if (dataArrayToken == null || dataArrayToken.Type != JTokenType.Array)

throw new InvalidOperationException("Nessun 'data_array' trovato nel JSON.");

var dataArray = (JArray)dataArrayToken;

// 2) Prova a leggere lo schema colonne se presente

var columnsToken = jsonObject.SelectToken("manifest.schema.columns") as JArray;

var columnNames = new List<string>();

int maxCols;

if (columnsToken != null && columnsToken.Count > 0)

{

foreach (var col in columnsToken)

{

// Cerca proprietà comuni per il nome colonna

var name = (string)(col["name"] ?? col["title"] ?? col["id"]);

if (string.IsNullOrWhiteSpace(name))

name = $"Col{columnNames.Count + 1}";

columnNames.Add(name);

}

maxCols = columnNames.Count;

}

else

{

// 3) Nessuno schema: ricava il numero max di colonne dalle righe

maxCols = dataArray

.Select(r => (r as JArray)?.Count ?? 0)

.DefaultIfEmpty(0)

.Max();

for (int i = 1; i <= maxCols; i++)

columnNames.Add($"Col{i}");

}

// 4) Crea il DataTable e popola i dati

var table = new DataTable();

foreach (var name in columnNames)

table.Columns.Add(name, typeof(string));

foreach (var rowToken in dataArray)

{

var rowArray = rowToken as JArray;

if (rowArray == null)

continue; // oppure: throw, se vuoi essere più rigido

var dr = table.NewRow();

for (int i = 0; i < maxCols; i++)

{

if (i < rowArray.Count)

{

var cell = rowArray[i];

// Se vuoi stringa vuota al posto del DBNull:

// dr[i] = cell.Type == JTokenType.Null ? "" : (string)cell;

dr[i] = cell.Type == JTokenType.Null ? (object)DBNull.Value : (string)cell;

}

else

{

dr[i] = DBNull.Value;

}

table.Rows.Add(dr);

}

return table;

}

public static string EscapeForJson(string s)

{

if (s == null) return "";

return s.Replace("\\", "\\\\").Replace("\"", "\\\"");

}

public static (string StatementId, string State) ParseWithNewtonsoft(string json)

{

var jo = JObject.Parse(json);

string statementId = (string)jo["statement_id"];

string state = (string)jo["status"]?["state"];

return (statementId, state);

}

]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

</inputs>

</stage>

</outputs>

</stage>

<narrative>Clean Up Page

This is an optional page where you might choose to perform some finalisation (or "cleanup") tasks as your business object is closed down.

The cleanup action will be called automatically immediately after closing your business object at the end of a business process.

You will not be able to call this action from a business process, nor will it be called at any other time than before the disposal of the business object.</narrative>

</stage>

<narrative>Initialise Page

This is an optional page where you might choose to perform some initialisation tasks after your business object is loaded.

The initialise action will be called automatically immediately after loading your business object.

You will not be able to call this action from a business process, nor will it be called at any other time than after the creation of the object.</narrative>

</stage>

</inputs>

</outputs>

<code><![CDATA[

// ======================

// OUTPUT atteso

// ======================

Success = false;

ErrorMessage = "";

JSONOutput = "";

// ======================

// Impostazioni

// ======================

// Disposition: "INLINE" (data_array) o "EXTERNAL_LINKS" (URL SAS)

string disposition = "INLINE";

// Formato risultati: con INLINE supportato "JSON_ARRAY"

string format = "JSON_ARRAY";

// Polling

TimeSpan pollingTimeout = TimeSpan.FromMinutes(10);

int initialDelayMs = 500;

int maxDelayMs = 5000;

// ======================

// Helper errori/estrazione

// ======================

string ExtractDetail(JObject j)

{

if (j == null) return null;

return (string)j["status"]?["error"]?["message"]

?? (string)j["error"]?["message"]

?? (string)j["status"]?["error"]?["code"]

?? (string)j["error"]?["code"];

}

string ExtractSqlState(JObject j)

{

if (j == null) return null;

return (string)j["status"]?["error"]?["sql_state"]

?? (string)j["error"]?["sql_state"];

}

void BuildShortErrorFromHttp(string forcedState, HttpResponseMessage resp, string body, out string errorMessage)

{

var stateLabel = string.IsNullOrWhiteSpace(forcedState) ? "FAILED" : forcedState;

string detail = null;

string sqlState = null;

try

{

var j = string.IsNullOrWhiteSpace(body) ? null : JObject.Parse(body);

detail = ExtractDetail(j);

sqlState = ExtractSqlState(j);

}

catch { /* ignore parse errors */ }

if (string.IsNullOrEmpty(detail))

detail = $"HTTP {(int)resp.StatusCode} {resp.ReasonPhrase}";

if (string.IsNullOrEmpty(sqlState))

sqlState = "N/A";

errorMessage = $"State: {stateLabel} - Detail: {detail} - SQLSTATE: {sqlState}";

}

// ======================

// Fetch di un CHUNK (INLINE o EXTERNAL_LINKS) → ritorna JArray righe del chunk

// ======================

(bool ok, JArray rows, string err) FetchChunkRows(

HttpClient clientWithAuth,

string apiBaseUrl,

string token,

string statementId,

int chunkIndex)

{

try

{

var url = $"{apiBaseUrl.TrimEnd('/')}/{statementId}/result/chunks/{chunkIndex}";

var req = new HttpRequestMessage(HttpMethod.Get, url);

req.Headers.Add("Authorization", $"Bearer {token}");

var resp = clientWithAuth.SendAsync(req).Result;

var body = resp.Content.ReadAsStringAsync().Result;

if (!resp.IsSuccessStatusCode)

{

BuildShortErrorFromHttp("FAILED", resp, body, out var msg);

return (false, null, msg);

}

var chunk = JObject.Parse(body);

// Caso INLINE: data_array in root

if (chunk["data_array"] is JArray inlineRows)

{

return (true, inlineRows, null);

}

// Caso EXTERNAL_LINKS: scarico i link SAS (ognuno è un JSON array)

if (chunk["external_links"] is JArray links)

{

var merged = new JArray();

foreach (var l in links)

{

var link = (string)l["external_link"];

if (string.IsNullOrEmpty(link)) continue;

using (var http = new HttpClient()) // NO Authorization header per SAS

{

var dl = http.GetStringAsync(link).Result; // contenuto JSON_ARRAY

var jarr = JArray.Parse(dl);

foreach (var r in jarr) merged.Add(r);

// libera presto memoria intermedia

jarr.Clear();

GC.Collect();

GC.WaitForPendingFinalizers();

}

return (true, merged, null);

}

return (false, null, "State: FAILED - Detail: Nessun 'data_array' o 'external_links' nel chunk. - SQLSTATE: N/A");

}

catch (Exception ex)

{

return (false, null, $"State: FAILED - Detail: {ex.Message} - SQLSTATE: N/A");

}

// ======================

// MAIN FLOW

// ======================

HttpClient client = new HttpClient();

try

{

// ====== POST (asincrono): wait_timeout=0s → ritorna subito statement_id ======

var payload = new

{

statement = sqlQuery,

warehouse_id = warehouseId,

wait_timeout = "0s",

disposition = disposition,

format = format

};

var content = new StringContent(

JsonConvert.SerializeObject(payload),

Encoding.UTF8,

"application/json"

);

var postReq = new HttpRequestMessage(HttpMethod.Post, baseUrl);

postReq.Headers.Add("Authorization", $"Bearer {accessToken}");

postReq.Content = content;

var postResp = client.SendAsync(postReq).Result;

var postBody = postResp.Content.ReadAsStringAsync().Result;

if (!postResp.IsSuccessStatusCode)

{

BuildShortErrorFromHttp("FAILED", postResp, postBody, out ErrorMessage);

Success = false;

return;

}

JObject jo = null;

try { jo = JObject.Parse(postBody); } catch { }

if (jo == null)

{

Success = false;

ErrorMessage = "State: FAILED - Detail: Impossibile interpretare la risposta POST come JSON. - SQLSTATE: N/A";

return;

}

var statementId = (string)jo["statement_id"];

var state = (string)jo["status"]?["state"];

// Stati terminali previsti

string[] terminal = { "SUCCEEDED", "FAILED", "CANCELED", "CLOSED" };

bool needsPolling =

!string.IsNullOrWhiteSpace(statementId) &&

(string.IsNullOrWhiteSpace(state) || Array.IndexOf(terminal, state?.ToUpperInvariant()) < 0);

// ====== POLLING ======

JObject finalObj = jo;

if (needsPolling)

{

DateTime start = DateTime.UtcNow;

int delayMs = initialDelayMs;

string getUrl = baseUrl.TrimEnd('/') + "/" + statementId;

while (true)

{

if (DateTime.UtcNow - start > pollingTimeout)

{

Success = false;

ErrorMessage = "State: FAILED - Detail: Polling scaduto lato client. - SQLSTATE: N/A";

return;

}

Thread.Sleep(delayMs);

delayMs = Math.Min(delayMs * 2, maxDelayMs);

var pollReq = new HttpRequestMessage(HttpMethod.Get, getUrl);

pollReq.Headers.Add("Authorization", $"Bearer {accessToken}");

var pollResp = client.SendAsync(pollReq).Result;

var pollBody = pollResp.Content.ReadAsStringAsync().Result;

if (pollResp.StatusCode == HttpStatusCode.NotFound)

{

Success = false;

ErrorMessage = "State: FAILED - Detail: Statement non trovato o scaduto. - SQLSTATE: N/A";

return;

}

if (!pollResp.IsSuccessStatusCode)

{

BuildShortErrorFromHttp("FAILED", pollResp, pollBody, out ErrorMessage);

Success = false;

return;

}

JObject p = null;

try { p = JObject.Parse(pollBody); } catch { p = null; }

if (p == null) continue;

finalObj = p;

state = (string)p["status"]?["state"];

if (string.IsNullOrWhiteSpace(state)) continue;

var upper = state.ToUpperInvariant();

if (Array.IndexOf(terminal, upper) >= 0)

{

if (upper != "SUCCEEDED")

{

Success = false;

var detail = ExtractDetail(p) ?? $"State={upper}";

var sqlState = ExtractSqlState(p) ?? "N/A";

ErrorMessage = $"State: {upper} - Detail: {detail} - SQLSTATE: {sqlState}";

return;

}

break; // SUCCEEDED

}

// ====== STATO: SUCCEEDED → costruzione JSONOutput in streaming ======

int totalChunks = (int?)finalObj["manifest"]?["total_chunk_count"] ?? 1;

// Builder per l'output finale (evita JArray globale in RAM)

var sb = new StringBuilder(1024 * 1024); // 1MB iniziale; cresce se serve

// Header JSON (mettiamo row_count dopo aver scritto tutte le righe)

sb.Append("{\"statement_id\":\"")

.Append((string)finalObj["statement_id"])

.Append("\",\"status\":\"SUCCEEDED\",\"manifest\":{\"total_chunk_count\":")

.Append(totalChunks)

.Append("},\"result\":{\"chunk_index\":0,\"row_offset\":0,\"data_array\":[");

bool firstRow = true;

long rowCount = 0;

// Funzione locale: appende righe e libera memoria del chunk

void AppendRows(JArray rows)

{

foreach (var r in rows)

{

if (!firstRow) sb.Append(',');

sb.Append(r.ToString(Newtonsoft.Json.Formatting.None));

firstRow = false;

rowCount++;

}

// libera memoria del chunk

rows.Clear();

GC.Collect();

GC.WaitForPendingFinalizers();

}

// 1) Se il polling ha fornito già il primo chunk in finalObj.result.data_array → usalo

if (finalObj["result"]?["data_array"] is JArray chunk0)

{

AppendRows(chunk0);

}

else

{

// Altrimenti: scarica il chunk 0

var (ok0, rows0, err0) = FetchChunkRows(client, baseUrl, accessToken, statementId, 0);

if (!ok0)

{

Success = false;

ErrorMessage = err0;

return;

}

AppendRows(rows0);

}

// 2) Scarica e appende i chunk successivi (1..N-1)

for (int i = 1; i < totalChunks; i++)

{

var (ok, rows, err) = FetchChunkRows(client, baseUrl, accessToken, statementId, i);

if (!ok)

{

Success = false;

ErrorMessage = err;

return;

}

AppendRows(rows);

}

// Chiudi array e inserisci row_count

sb.Append("],\"row_count\":")

.Append(rowCount)

.Append("}}");

// Assegna alla variabile di output richiesta

JSONOutput = sb.ToString();

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = $"State: FAILED - Detail: {ex.Message} - SQLSTATE: N/A";

}

finally

{

client?.Dispose();

}

]]></code>

</stage>

<initialvalue xml:space="preserve">https://westeurope.azuredatabricks.net/api/2.0/sql/statements</initialvalue>

</stage>

</stage>

</stage>

</stage>

<initialvalue>False</initialvalue>

</stage>

</stage>

</stage>

<datatype>collection</datatype>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

dataTable = null;

dataTable = new DataTable();

try

{

dataTable = GenerateDataTableFromJson(json);

Success = true;

GC.Collect();

GC.WaitForPendingFinalizers();

}

catch (Exception ex)

{

Success = false;

ErrorMessage = ex.Message;

GC.Collect();

GC.WaitForPendingFinalizers();

}

]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

<initialvalue>False</initialvalue>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

try

{

DataTableToExcel exporter = new DataTableToExcel();

exporter.ExportToExcel(DatatableResult, Filepath, SheetName);

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = ex.Message;

}]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</outputs>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

<datatype>password</datatype>

</initialvalueenc>

</stage>

</inputs>

</outputs>

<code><![CDATA[

string tokenUrl = $"https://login.microsoftonline.com/{TenantID}/oauth2/v2.0/token";

using (var httpClient = new HttpClient())

{

var form = new FormUrlEncodedContent(new[]

{

new KeyValuePair<string, string>("client_id", ClientId),

new KeyValuePair<string, string>("client_secret", ClientSecret),

new KeyValuePair<string, string>("grant_type", "client_credentials"),

// Scope corretto per Azure Databricks (AAD resource + /.default)

new KeyValuePair<string, string>("scope", "2ff814a6-3304-4ab8-85cb-cd0e6f879c1d/.default")

});

var resp = httpClient.PostAsync(tokenUrl, form).GetAwaiter().GetResult();

string body = resp.Content.ReadAsStringAsync().GetAwaiter().GetResult();

if (!resp.IsSuccessStatusCode)

{

throw new Exception($"Token request failed ({resp.StatusCode}): {body}");

}

var json = JObject.Parse(body);

string accessToken = json["access_token"]?.ToString();

if (string.IsNullOrEmpty(accessToken))

throw new Exception("access_token missing in token response");

AccessToken = accessToken;

}

]]></code>

</stage>

</stage>

</stage>

</inputs>

</stage>

</outputs>

</stage>

</stage>

</stage>

</stage>

<initialvalue>False</initialvalue>

</stage>

</stage>

</stage>

<datatype>collection</datatype>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

dataTable = null;

dataTable = new DataTable();

try

{

dataTable = GenerateDataTableFromJson(json);

Success = true;

GC.Collect();

GC.WaitForPendingFinalizers();

}

catch (Exception ex)

{

Success = false;

ErrorMessage = ex.Message;

GC.Collect();

GC.WaitForPendingFinalizers();

}

]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

<initialvalue>False</initialvalue>

</stage>

</inputs>

</outputs>

<code><![CDATA[Success = false;

ErrorMessage = "";

try

{

DataTableToExcel exporter = new DataTableToExcel();

exporter.ExportToExcel(DatatableResult, Filepath, SheetName);

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = ex.Message;

}]]></code>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

<onsuccess>0f46bfec-9fc3-4b45-9073-b2a2feebbbe3</onsuccess>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</stage>

</inputs>

</outputs>

<code><![CDATA[

// Assumo che tu abbia già queste variabili in scope:

// string databricksUrl; // es: "https://<workspace>/api/2.0/sql/statements"

// string accessToken; // Bearer token del Service Principal

// string sqlQuery;

// string warehouseId;

Success = false;

ErrorMessage = "";

JSONOutput = "";

// == Impostazione opzionale ==

bool fetchAllChunks = true; // true = unisce tutti i chunk; false = solo chunk 0

TimeSpan overallTimeout = TimeSpan.FromMinutes(10);

int delayMs = 500;

int maxDelayMs = 5000;

// --- Helper locali identici ai tuoi ---

string ExtractDetail(JObject j)

{

if (j == null) return null;

return (string)j["status"]?["error"]?["message"]

?? (string)j["error"]?["message"]

?? (string)j["status"]?["error"]?["code"]

?? (string)j["error"]?["code"];

}

string ExtractSqlState(JObject j)

{

if (j == null) return null;

return (string)j["status"]?["error"]?["sql_state"]

?? (string)j["error"]?["sql_state"];

}

void BuildShortErrorFromHttp(string forcedState, HttpResponseMessage resp, string body, out string errorMessage)

{

var stateLabel = string.IsNullOrWhiteSpace(forcedState) ? "FAILED" : forcedState;

string detail = null;

string sqlState = null;

try

{

var j = string.IsNullOrWhiteSpace(body) ? null : JObject.Parse(body);

detail = ExtractDetail(j);

sqlState = ExtractSqlState(j);

}

catch { /* ignore parse errors */ }

if (string.IsNullOrEmpty(detail))

detail = $"HTTP {(int)resp.StatusCode} {resp.ReasonPhrase}";

if (string.IsNullOrEmpty(sqlState))

sqlState = "N/A";

errorMessage = $"State: {stateLabel} - Detail: {detail} - SQLSTATE: {sqlState}";

}

// --- Funzioni locali per i chunk ---

(bool ok, JObject resultObj, string err) FetchAndMergeChunks(

HttpClient httpClient,

string baseUrl,

string statementId,

int totalChunks,

bool fetchAll)

{

try

{

var allRows = new JArray();

int chunksToFetch = fetchAll ? totalChunks : Math.Min(1, totalChunks);

for (int i = 0; i < chunksToFetch; i++)

{

var url = $"{baseUrl.TrimEnd('/')}/{statementId}/result?chunk_index={i}";

var resp = httpClient.GetAsync(url).GetAwaiter().GetResult();

var body = resp.Content.ReadAsStringAsync().GetAwaiter().GetResult();

if (!resp.IsSuccessStatusCode)

{

BuildShortErrorFromHttp("FAILED", resp, body, out var errMsg);

return (false, null, errMsg);

}

JObject chunk = null;

try { chunk = JObject.Parse(body); } catch { }

if (chunk == null)

{

return (false, null, "State: FAILED - Detail: Impossibile interpretare la risposta dei risultati come JSON. - SQLSTATE: N/A");

}

var rows = (JArray)chunk["data_array"] ?? new JArray();

foreach (var r in rows) allRows.Add(r);

}

var resultObj = new JObject

{

["chunk_index"] = 0,

["row_offset"] = 0,

["row_count"] = allRows.Count,

["data_array"] = allRows

};

return (true, resultObj, null);

}

catch (Exception ex)

{

return (false, null, $"State: FAILED - Detail: {ex.Message} - SQLSTATE: N/A");

}

try

{

using (var httpClient = new HttpClient())

{

httpClient.DefaultRequestHeaders.Add("Authorization", $"Bearer {accessToken}");

// 1) POST: crea statement (senza attendere)

var payload = new

{

statement = sqlQuery,

warehouse_id = warehouseId,

wait_timeout = "0s",

format = "JSON_ARRAY" // così hai data_array coerente col primo metodo

};

var content = new StringContent(

JsonConvert.SerializeObject(payload),

Encoding.UTF8,

"application/json"

);

var postResp = httpClient.PostAsync(databricksUrl, content).GetAwaiter().GetResult();

var postBody = postResp.Content.ReadAsStringAsync().GetAwaiter().GetResult();

if (!postResp.IsSuccessStatusCode)

{

BuildShortErrorFromHttp("FAILED", postResp, postBody, out ErrorMessage);

Success = false;

return;

}

JObject postJson = null;

try { postJson = JObject.Parse(postBody); } catch { }

if (postJson == null)

{

ErrorMessage = "State: FAILED - Detail: Impossibile interpretare la risposta POST come JSON. - SQLSTATE: N/A";

Success = false;

return;

}

var statementId = (string)postJson["statement_id"];

var stateImmediate = (string)postJson["status"]?["state"];

if (string.IsNullOrEmpty(statementId))

{

// Caso finale immediato

if (!"SUCCEEDED".Equals(stateImmediate, StringComparison.OrdinalIgnoreCase))

{

var detail = ExtractDetail(postJson) ?? "N/A";

var sqlState = ExtractSqlState(postJson) ?? "N/A";

ErrorMessage = $"State: {stateImmediate ?? "FAILED"} - Detail: {detail} - SQLSTATE: {sqlState}";

Success = false;

return;

}

else

{

// Già completo → restituisco l'oggetto come nel primo metodo

JSONOutput = postJson.ToString(Formatting.None);

Success = true;

return;

}

// 2) Polling fino a stato terminale

var start = DateTime.UtcNow;

string finalState = null;

JObject finalStatement = null;

string[] terminal = { "SUCCEEDED", "FAILED", "CANCELED", "CLOSED" };

while (true)

{

var getUrl = $"{databricksUrl.TrimEnd('/')}/{statementId}";

var getResp = httpClient.GetAsync(getUrl).GetAwaiter().GetResult();

var getBody = getResp.Content.ReadAsStringAsync().GetAwaiter().GetResult();

if (!getResp.IsSuccessStatusCode)

{

BuildShortErrorFromHttp("FAILED", getResp, getBody, out ErrorMessage);

Success = false;

return;

}

JObject getJson = null;

try { getJson = JObject.Parse(getBody); } catch { }

if (getJson == null)

{

ErrorMessage = "State: FAILED - Detail: Impossibile interpretare la risposta GET polling come JSON. - SQLSTATE: N/A";

Success = false;

return;

}

finalStatement = getJson;

var state = (string)getJson["status"]?["state"];

if (!string.IsNullOrWhiteSpace(state) && terminal.Contains(state.ToUpperInvariant()))

{

finalState = state;

break;

}

if (DateTime.UtcNow - start > overallTimeout)

{

ErrorMessage = $"State: FAILED - Detail: Polling scaduto dopo {overallTimeout.TotalSeconds:N0} secondi. - SQLSTATE: N/A";

Success = false;

return;

}

Thread.Sleep(delayMs);

delayMs = Math.Min(delayMs * 2, maxDelayMs);

}

// 3) Esito finale

if (!"SUCCEEDED".Equals(finalState, StringComparison.OrdinalIgnoreCase))

{

var detail = ExtractDetail(finalStatement) ?? "N/A";

var sqlState = ExtractSqlState(finalStatement) ?? "N/A";

ErrorMessage = $"State: {finalState} - Detail: {detail} - SQLSTATE: {sqlState}";

Success = false;

return;

}

// 4) Recupero risultati e innesto in finalStatement

JToken resultToken = finalStatement["result"];

// Quanti chunk ci sono?

int totalChunks = 1;

try

{

totalChunks = (int?)finalStatement["manifest"]?["total_chunk_count"] ?? 1;

}

catch { /* ignore */ }

if (resultToken == null)

{

// Non c'è result nello statement → scarico i chunk e UNISCO

var (ok, resultObj, err) = FetchAndMergeChunks(httpClient, databricksUrl, statementId, totalChunks, fetchAllChunks);

if (!ok)

{

ErrorMessage = err;

Success = false;

return;

}

finalStatement["result"] = resultObj;

}

else

{

// C'è già result nello statement.

// Se vuoi essere sicuro di avere TUTTO, e ci sono più chunk, li recupero e unisco.

if (fetchAllChunks && totalChunks > 1)

{

// Riparto da zero e unisco tutti i chunk (più semplice e robusto)

var (ok, resultObj, err) = FetchAndMergeChunks(httpClient, databricksUrl, statementId, totalChunks, fetchAllChunks);

if (!ok)

{

ErrorMessage = err;

Success = false;

return;

}

finalStatement["result"] = resultObj;

}

// Allineo counters (opzionale)

try

{

var dataArray = finalStatement["result"]?["data_array"] as JArray;

if (dataArray != null)

{

finalStatement["result"]["row_count"] = dataArray.Count;

finalStatement["result"]["chunk_index"] = 0;

finalStatement["result"]["row_offset"] = 0;

}

catch { /* ignore */ }

// 5) OUTPUT: restituisco lo statement completo (come nel primo metodo)

JSONOutput = finalStatement.ToString(Formatting.None);

Success = true;

}

catch (Exception ex)

{

Success = false;

ErrorMessage = $"State: FAILED - Detail: {ex.Message} - SQLSTATE: N/A";

}

]]></code>

</stage>

<narrative>https://adb-<workspace-id>.<region-shard>.azuredatabricks.net/api/2.0/sql/statements</narrative>

</stage>

</outputs>

</stage>

</process>

SS&C Blue Prism Community

Databricks API JSON Result to Collection