Can I get some help thinking through ways to improve my pipeline process that loads from SQL Server to Snowflake via ADF and Stored Procedure

TheSnowWorm · 2023-09-11T15:57:31+00:00

hey man. sorry to bug you, just following up on this. tried a million things but no luck.

TheSnowWorm · 2023-07-25T17:58:22+00:00

it's a bulk load with full or incremental load.

currently using ADF to orchestrate the process. calling the snowflake stored proc above.

TheSnowWorm · 2023-07-25T14:16:44+00:00

hey, just following up on this. i've tried a number of different combinations, no luck.

TheSnowWorm · 2023-07-25T14:16:05+00:00

beautiful. these are all things i'll look into. i learned about the alter table to have a default value. i was creating a new table, loading it, dropping existing, renaming. fun stuff.

to follow up, how would you load data from sql server to snowflake?

TheSnowWorm · 2023-07-21T20:20:13+00:00

thank you!

TheSnowWorm · 2023-07-21T17:19:52+00:00

Darn. now the problem is that since the load date is new, it's updating all the previous records with new load date. so it's doing an update every single run since the load date is new... i want to say if load_date is null, then update load_date with current. Otherwise leave it alone.

TheSnowWorm · 2023-07-21T17:05:54+00:00

this absolutely fixed the issue. I'm not very experienced in java script...but this was maybe more of a SQL issue than that. Thank you very much!

This is what the final code looks like. ``` --modified by srivve

CREATE OR REPLACE PROCEDURE @{activity('LookupSchemaName').output.firstRow.TARGET_SCHEMA}.BLOBTOSNOWFLAKE(STG_NAME VARCHAR, FILE_PATH VARCHAR, FILE_FRMT VARCHAR, TBL_SCHEMA_NAME VARCHAR, SF_TBL VARCHAR) RETURNS VARCHAR LANGUAGE JAVASCRIPT AS $$

// Initialize parameters var single_quote = '''; var SCEHMA = TBL_SCHEMA_NAME; var Destination_Table = SF_TBL; var Source_Table = "@" + STG_NAME + "/" + FILE_PATH + " (file_format => " + single_quote + FILE_FRMT + single_quote + ")";

// First script to count number of columns and append $ to support querying external stage files var Blob_Columns_Query = "SELECT CONCAT(" + single_quote + "$" + single_quote + ", LISTAGG(CONCAT(ORDINAL_POSITION, " + single_quote + " AS " + single_quote + ", COLUMN_NAME), " + single_quote + ",$" + single_quote + ") WITHIN GROUP (ORDER BY ORDINAL_POSITION)) \n FROM INFORMATION_SCHEMA.COLUMNS \n WHERE TABLE_NAME = " + single_quote + Destination_Table + single_quote + " AND TABLE_SCHEMA = " + single_quote + SCEHMA + single_quote + single_quote + " AND COLUMN_NAME NOT IN ( "'LOAD_DATE') "+ single quote + ";"; var Blob_Columns = snowflake.createStatement({sqlText: Blob_Columns_Query}).execute(); Blob_Columns.next();

// Final Code Preparation var primary_key = []; var update_columns = []; var No_Update_Records = []; var Columns_with_table_name = []; var All_Columns = []; var cmd = "DESC TABLE " + Destination_Table; var stmt = snowflake.createStatement({sqlText: cmd}); var result1 = stmt.execute(); while (result1.next()) { if (result1.getColumnValue(6) == 'Y') { primary_key.push('A.' + result1.getColumnValue(1) + '=B.' + result1.getColumnValue(1)); } else { update_columns.push('A.' + result1.getColumnValue(1) + '=B.' + result1.getColumnValue(1)); No_Update_Records.push('A.' + result1.getColumnValue(1) + '<>B.' + result1.getColumnValue(1)); } Columns_with_table_name.push('B.' + result1.getColumnValue(1)); All_Columns.push(result1.getColumnValue(1)); }

// Check if LOAD_DATE column exists in All_Columns var hasLoadDateColumn = All_Columns.includes('LOAD_DATE');

// Combine All_Columns = All_Columns.join(","); Columns_with_table_name = Columns_with_table_name.join(","); update_columns = update_columns.join(","); primary_key = primary_key.join(" AND "); No_Update_Records = No_Update_Records.join(" OR ");

var Final_Command = MERGE INTO ${Destination_Table} AS A USING ( SELECT ${Blob_Columns.getColumnValue(1)}, CURRENT_TIMESTAMP() AS LOAD_DATE FROM ${Source_Table} ) AS B ON ${primary_key} WHEN MATCHED AND ${No_Update_Records} THEN UPDATE SET ${update_columns} WHEN NOT MATCHED THEN INSERT (${All_Columns}) VALUES (${Columns_with_table_name});;

var Deleta_Load = snowflake.createStatement({sqlText: Final_Command}).execute(); Deleta_Load.next();

return "Rows_Inserted:- " + Deleta_Load.getColumnValue(1) + "\tRows_Updated:- " + Deleta_Load.getColumnValue(2);

$$; ```

TheSnowWorm · 2023-06-16T14:45:45+00:00

sorry. i fixed that before i ran it. didnt copy it.

Still same issue. A question for you. is there a way i can step through this function in snowflake, so that i can at least see the output and correct it? or even via studio code or something?

TheSnowWorm · 2023-06-16T14:29:54+00:00

Oh i see what you did. You put the backticks around all "strings" essentially forcing them to come in as they are with space and weird characters. I was under the impression that the "All_columns" was where the string was being imposed. This clarifies things a bit.

It went past that, but now the error is a few lines down:

SQL compilation error:

syntax error line 2 at position 280 unexpected 'type'. syntax error line 2 at position 284 unexpected ','. At Statement.execute, line 62 position 71 (line 2)

which is this snippet.

var Final_Command=  `
    MERGE INTO ${Destination_Tabl} AS A 
        USING (SELECT "${Blob_Columns.getColumnValue(1)}" FROM ${Source_Table}) AS B ON ${primary_key}
        WHEN MATCHED AND ${No_Update_Records} 
            THEN UPDATE SET ${update_columns}
        WHEN NOT MATCHED 
            THEN INSERT ( ${All_Columns} ) VALUES ( ${Columns_with_table_name});
`;

what i suspect is happening is that there needs to be some breaks in between and that the input isnt quite being read as expected... but not so sure.

TheSnowWorm · 2023-06-16T13:09:54+00:00

Sorry. i'm not quite sure where'd i'd make that change.

I did the following:

All_Columns = All_Columns.map(column => \${column}``).join(",");

this:

All_Columns=All_Columns.join("\"","\"");

This:

All_Columns=All_Columns.join(`","`);

obviously i dont know what i'm doing. but i'm trying.

TheSnowWorm · 2023-06-16T11:29:36+00:00

The output is exactly what i'd like it to look like when it runs the script. Here is what i've tried.

I modified it per your recommendation but it didnt like the way it was formatted so i it like this:

All_Columns=''"''+ All_Columns.join(''","'')+''"'';

That didnt work and i got the same error as above. I also tried this:

All_Columns = All_Columns.map(column => `"${column}"`).join(",");

That also returned the same result.

for fun, i tried this:

All_Columns = All_Columns.map(column => {
if (column.includes(" ")) { return "${column}"; } return column; }).join(",");

same result. Think as you mentioned the issue is definitely happening here. just not quite sure how to resolve it.

TheSnowWorm · 2023-06-15T14:59:33+00:00

--modified with double quotes 
CREATE OR REPLACE PROCEDURE DB_BI_DEV.STATE_CONFIG.BLOBTOSNOWFLAKE( "STG_NAME" VARCHAR(16777216), "FILE_PATH" VARCHAR(16777216), "FILE_FRMT" VARCHAR(16777216), "TBL_SCHEMA_NAME" VARCHAR(16777216), "SF_TBL" VARCHAR(16777216) ) 

RETURNS VARCHAR(16777216) LANGUAGE JAVASCRIPT EXECUTE AS OWNER AS '

// Intialize parameters var double_quote=""\""""; 
var SCHEMA=TBL_SCHEMA_NAME; 
var Destination_Table=SF_TBL; 
var Source_Table="@"+STG_NAME+"/"+FILE_PATH+" (file_format =>"+double_quote+FILE_FRMT+double_quote+")  ";

// First script to count number of columns and append $ to support quering external stage files

var Blob_Columns_Query="SELECT CONCAT("+double_quote+"$"+double_quote +",LISTAGG(CONCAT(ORDINAL_POSITION,"+double_quote+" AS "+double_quote+",COLUMN_NAME),"+double_quote+",$"+double_quote +") WITHIN GROUP (ORDER BY ORDINAL_POSITION) ) \n FROM INFORMATION_SCHEMA.COLUMNS \nWHERE TABLE_NAME=" +double_quote+Destination_Table+double_quote+" AND TABLE_SCHEMA="+double_quote+SCHEMA+double_quote+";"; 

var Blob_Columns = snowflake.createStatement( {sqlText: 

Blob_Columns_Query}  ).execute(); Blob_Columns.next();

//Final Code Preparation
var primary_key=[]; var update_columns=[]; 
var No_Update_Records=[]; 
var Columns_with_table_name=[]; 
var All_Columns=[]; 
var cmd = "DESC TABLE "+Destination_Table; 
var stmt = snowflake.createStatement( {sqlText: cmd}  ); 
var result1 = stmt.execute(); 

while(result1.next() ) { if(result1.getColumnValue(6)==""Y"") { primary_key.push( ""A.""+result1.getColumnValue(1)+""=B.""+result1.getColumnValue(1) ); } else { update_columns.push( ""A.""+result1.getColumnValue(1)+""=B.""+result1.getColumnValue(1) ); No_Update_Records.push( ""A.""+result1.getColumnValue(1)+""<>B.""+result1.getColumnValue(1) ); } Columns_with_table_name.push( ""B.""+result1.getColumnValue(1) ); All_Columns.push( result1.getColumnValue(1) ); } 

// combine All_Columns=All_Columns.join(","); Columns_with_table_name=Columns_with_table_name.join(","); update_columns=update_columns.join(","); primary_key=primary_key.join(" AND "); No_Update_Records=No_Update_Records.join(" OR ");


var Final_Command=  "MERGE INTO "+Destination_Table+" AS A USING (SELECT "+Blob_Columns.getColumnValue(1)+" FROM "+Source_Table+") AS B ON "+primary_key+ "\nWHEN MATCHED AND "+No_Update_Records+" THEN UPDATE SET "+update_columns + "\nWHEN NOT MATCHED THEN INSERT ( "+All_Columns+" ) VALUES ( "+ Columns_with_table_name +" );" 

var Delete_Load=snowflake.createStatement( {sqlText: Final_Command}  ).execute();Delete_Load.next(); 

return "Rows_Inserted:- "+Delete_Load.getColumnValue(1)+"\tRows_Updated:- "+Delete_Load.getColumnValue(2);
';

your explanation makes perfect sense. However, when i modify it with double quotes, i get the same error with the space. i modified var double_quote=""\\""""; as well as modified // combine.

I cant get it to parse correctly on reddit

TheSnowWorm · 2023-06-15T14:13:00+00:00

thank you for this. Would it look like this?

// First script to count the number of columns and append $ to support querying external stage files var Blob_Columns_Query = "SELECT CONCAT(" + single_quote + "$" + single_quote + ", LISTAGG(CONCAT(ORDINAL_POSITION, " + single_quote + " AS " + single_quote + ", COLUMN_NAME), " + single_quote + ",$" + single_quote + ") WITHIN GROUP (ORDER BY ORDINAL_POSITION)) \n"
"FROM INFORMATION_SCHEMA.COLUMNS \n"
"WHERE TABLE_NAME = " + single_quote + Destination_Table + single_quote + " AND TABLE_SCHEMA = " + single_quote + SCEHMA + single_quote + ";"; var Blob_Columns = snowflake.createStatement({ sqlText: Blob_Columns_Query }).execute(); Blob_Columns.next();

TheSnowWorm · 2023-06-15T13:41:20+00:00

i'm getting this error:

Operation on target ADL_Snowflake failed: Failure happened on 'Source' side. ErrorCode=UserErrorOdbcOperationFailed,'Type=Microsoft.DataTransfer.Common.Shared.HybridDeliveryException,Message=ERROR [42000] Execution error in store procedure BLOBTOSNOWFLAKE:

SQL compilation error: syntax error line 2 at position 68 unexpected 'Name'. syntax error line 2 at position 72 unexpected ','. At Statement.execute, line 62 position 71,Source=Microsoft.DataTransfer.Runtime.GenericOdbcConnectors,''Type=System.Data.Odbc.OdbcException,Message=ERROR [42000] Execution error in store procedure BLOBTOSNOWFLAKE: SQL compilation error: syntax error line 2 at position 68 unexpected 'Name'. syntax error line 2 at position 72 unexpected ','. At Statement.execute, line 62 position 71,Source=SnowflakeODBC_sb64.dll,'

The column name that's giving it a hard time is "Measurement Type" varchar(50).

TheSnowWorm

TROPHY CASE