Synapse notebook not working from pipeline

Question

Synapse notebook not working from pipeline

Blair 31

I have a Notebook in Azure Synapse written in spark (Scala).
It uses the synapsesql function to read data from a SQL pool and to push data into a SQL pool.

When I run the notebook everything works perfectly and the table in the SQL pool is created and populated. However when I call the notebook from a Synapse pipeline it succeeds but doesn't do anything

Any ideas?

Blair 31 Reputation points

2022-04-05T01:45:56.717+00:00

I have isolated the issue. I still don't have a way to resolve it.

When I turn off the firewall on the data lake it works successfully.

This isn't a resolution. Beyond whitelisting all the IR IP ranges, how do I fix this?
AnnuKumari-MSFT 27,926 Reputation points Microsoft Employee

2022-04-07T10:32:46.597+00:00

Hi @Blair ,

I have already escalated your issue to internal product team and conveyed your concern on finding other resolution apart from whitelisting IP . I will keep you posted once I hear regarding the fix.

Blair 31 Reputation points

2022-04-05T01:45:56.717+00:00

I have isolated the issue. I still don't have a way to resolve it.

When I turn off the firewall on the data lake it works successfully.

This isn't a resolution. Beyond whitelisting all the IR IP ranges, how do I fix this?
AnnuKumari-MSFT 27,926 Reputation points Microsoft Employee

2022-04-07T10:32:46.597+00:00

Hi @Blair ,

I have already escalated your issue to internal product team and conveyed your concern on finding other resolution apart from whitelisting IP . I will keep you posted once I hear regarding the fix.

Answer 1

AnnuKumari-MSFT 27,926 Microsoft Employee

Hi @Blair ,

Thankyou for using Microsoft Q&A platform and posting your query.

Looking at your query, it seems when you execute the notebook directly , you are able to see the output . However, when you call the same notebook via Synapse pipelines , pipeline gets succeeded but output doesn't show up .

It is an expected behavior. To view the output after pipeline execution ,

Pick up the Apache Spark application name from the output of pipeline
Navigate to Apache Spark Application under Monitor tab and search for the same application name .
These 4 tabs would be available there: Diagnostics,Logs,Input data,Output data
Go to Logs ad check 'stdout' for getting the required output.

Attaching video for your reference:

For more details, please refer to this video: https://www.youtube.com/watch?v=ydEXCVVGAiY

If this answers your query, please do let us know.

Blair 31 Reputation points

2022-03-29T22:00:34.67+00:00

This is all it says under stdout
AnnuKumari-MSFT 27,926 Reputation points Microsoft Employee

2022-03-30T15:22:03.06+00:00

Hi @Blair ,
I tried creating table using pyspark code in synapse notebook and called that notebook in my pipeline and executed it. It worked fine and created the table for me. Requesting you to kindly share the code which you are trying in your notebook. Thanks!
Blair 31 Reputation points

2022-03-30T23:26:34.667+00:00

%%spark

val pEnvironment = "t"
val pFolderName = "TestFolder"

val pSourceDatabaseName = "TestDatabase"
val pSourceSchemaName = "TestSchema"

val pRootFolderName = "RootFolder"
val pServerName = pEnvironment + "synas01"
val pDatabaseName = pEnvironment + "syndsqlp01"
val pTableName = pSourceDatabaseName + "" + pSourceSchemaName + "" + pFolderName

// Import functions and Synapse connector
import org.apache.spark.sql.DataFrame
import com.microsoft.spark.sqlanalytics.utils.Constants
import org.apache.spark.sql.functions._
import org.apache.spark.sql.SqlAnalyticsConnector._

// Get list of "FileLocation" from control.FileLoadStatus
val fls:DataFrame = spark.read.
synapsesql(s"${pDatabaseName}.control.FileLoadStatus").
select("FileLocation","ProcessedDate")

// Read all parquet files in folder into data frame
// Add file name as column
val df:DataFrame = spark.read.
parquet(s"/source/${pRootFolderName}/${pFolderName}/").
withColumn("FileLocation", input_file_name())

// Join parquet file data frame to FileLoadStatus data frame
// Exclude rows in parquet file data frame where ProcessedDate is not null
val df2 = df.
join(fls,Seq("FileLocation"), "left").
where(fls("ProcessedDate").isNull)

// Write data frame to sql table
df2.write.
option(Constants.SERVER,s"${pServerName}.sql.azuresynapse.net").
synapsesql(s"${pDatabaseName}.xtr.${pTableName}",Constants.INTERNAL)
AnnuKumari-MSFT 27,926 Reputation points Microsoft Employee

2022-03-31T13:48:14.323+00:00

@Blair ,
Thanks for sharing the code. I have reached out to the internal team for further investigation. Appreciate your patience ! Thanks
AnnuKumari-MSFT 27,926 Reputation points Microsoft Employee

2022-03-31T17:01:37.077+00:00

Hi @Blair ,

Could you please share the application logs? PG team responded saying there are multiple aspects in the shared code. Did you got a chance to triage for which part is failing or is successfully completing?
Blair 31 Reputation points

2022-04-01T00:37:51.037+00:00

The synapsesql lines throw a deprecated error but wont tell me what is deprecated, and I can't find anything online saying what might be deprecated with this command.

I added println commands to each step and they were all returned within the stdout output which tells me nothing has "failed"

I tried disabling the "val fls:DataFrame = spark.read." and "val df2 = df." sections so it just has the sections below and still no table created.

// Read all parquet files in folder into data frame
// Add file name as column
val df:DataFrame = spark.read.
parquet(s"/source/${pRootFolderName}/${pFolderName}/").
withColumn("FileLocation", input_file_name())
println("parquet dataframe populated")

// Write data frame to sql table
df.write.
option(Constants.SERVER,s"${pServerName}.sql.azuresynapse.net").
synapsesql(s"${pDatabaseName}.xtr.${pTableName}",Constants.INTERNAL)
println("table created and populated")

Under inputs I can see the parquet files.
Under outputs I get this error:
There has an BlobNotFound error be thrown, please check document to know more details.

Based on that I would say the error is likely in the last section with df.write. and for some reason when running the notebook through a pipeline it needs some access I haven't given it?
The data lake is associated with synapse and the synapse resource has contributor and storage blob contributor against the data lake.

Blair 31

This issue is still outstanding, except now the pipeline is failing. The notebook when run manually runs successfully.

Now when calling the Notebook through the pipeline the table is created but not populated and fails with the following error.

"errorCode": "6002",
"message": "com.microsoft.spark.sqlanalytics.SQLAnalyticsConnectorException: COPY statement input file schema discovery failed: Cannot bulk load. The file \"https://xxx.dfs.core.windows.net/landing/synapse/workspaces/xxx/sparkpools/xxx/sparkpoolinstances/6c183999-a58e-4295-bd6c-ac0bba99362c/livysessions/2022/04/05/188/tempdata/xxx/xtr/xxx/internal/ErrorIfExists/application_1649117340669_0001/1649117420550/part-00000-6febc423-9ab1-465c-8e3f-b923f501ac18-c000.snappy.parquet\" does not exist or you don't have file access rights.\n  at com.microsoft.spark.sqlanalytics.SqlAnalyticsConnectorClass$SQLAnalyticsFormatWriter.sqlanalytics(SqlAnalyticsConnectorClass.scala:324)\n\n  at com.microsoft.spark.sqlanalytics.SqlAnalyticsConnectorClass$SQLAnalyticsFormatWriter.synapsesql(SqlAnalyticsConnectorClass.scala:188)\n\n  ... 59 elided\n\nCaused by: com.microsoft.sqlserver.jdbc.SQLServerException: COPY statement input file schema discovery failed:

Has Microsoft changed something in the resource?

Synapse notebook not working from pipeline

1 answer

Activity