How do I create a dynamic SQL query for Cosmos DB using Azure Data Factory?

Andrew Hopkinson 6

I am doing a simple data flow where I have a source from a cosmos db container, I want to get the latest data that is greater than a _ts value that I store in a variable (SELECT * FROM c WHERE c_ts > @variables('LastRunDateTS')). How can I pass this to query in the source for the cosmos query? I see there are parameters, but I don't want to hard code the values, I would like these to come from a variable and it gives errors when I try and reference a variable inside the query. Please help, thanks.

phemanth 6,475 Reputation points Microsoft Vendor

2024-04-30T15:14:38.0433333+00:00
@Andrew Hopkinson
Thanks for using MS Q&A platform and posting your query.

You're on the right track! Cosmos DB supports parameterized queries, allowing you to use variables within your SQL statement. Here's how to achieve your goal:

1. Define the Variable:

The specific way to define variables depends on your data flow environment. However, it's generally done through a separate expression or script component before the Cosmos DB source.

2. Construct the Parameterized Query:

In your Cosmos DB source, use the SQL API syntax for parameterized queries. Here's the modified version of your original query:

SELECT * FROM c WHERE c_ts > @LastRunDateTS

Notice the @LastRunDateTS placeholder. This is where the variable value will be injected.

3. Pass the Variable Value:

In your data flow configuration, map the variable you defined earlier to the @LastRunDateTS parameter. This mapping might be done through a separate properties section or directly within the source component settings.

Here's a breakdown of potential issues and solutions:

Error referencing variable: Ensure you're using the correct syntax for referencing variables within your data flow environment. Check the documentation for your specific platform.

Incorrect mapping: Double-check that the variable you defined is correctly mapped to the @LastRunDateTS parameter in the Cosmos DB source configuration.

Consider using a type annotation for the variable to improve code clarity (e.g., string LastRunDateTS).

For complex variable logic, creating a separate script component before the Cosmos DB source might be helpful.

Hope this helps. Do let us know if you any further queries.
Andrew Hopkinson 6 Reputation points

2024-04-30T15:32:41.6933333+00:00

~~Thanks, but your SELECT statement is blank. Can you please fill in the select statement?~~ Oh it's there now.
Andrew Hopkinson 6 Reputation points

2024-04-30T15:43:47.02+00:00
How exactly do I do this? Can you please provide screen shots?

3. Pass the Variable Value:

In your data flow configuration, map the variable you defined earlier to the @LastRunDateTS parameter. This mapping might be done through a separate properties section or directly within the source component settings.

Also, to be clear, this has nothing to do with parameterizing the SQL statement in the source options?
Andrew Hopkinson 6 Reputation points

2024-04-30T16:00:11.7933333+00:00

I can't see a way to map the parameter to the variable. This gives an error.
Andrew Hopkinson 6 Reputation points

2024-04-30T16:01:19.3333333+00:00

I can't see a way to map the variable to the parameter
Andrew Hopkinson 6 Reputation points

2024-04-30T16:41:52.76+00:00
Ok, I think I figured some of it out, in the expression builder, this works

"select * from c where c._ts > 1710423660"

I create a parameter and map it to various variables, they all give errors

"select * from c where c._ts > $parameter1"

This validates in as an expression, but when I try and import the schema I get this:

{"innerErrorMessage":"null","cosmosDiagnostics":{"userAgent":"azsdk-java-cosmos/4.42.0-snapshot.1 Linux/4.15.0-1175-azure JRE/1.8.0_402 SparkConnector/azure-cosmos-spark_3-1_2-12/4.17.2|SYN|dfdebugpool-canadacentral|debugpool-8Cores-General-21|7976@vm-f5485468","activityId":"1ee27087-c5cb-4931-93e5-ce231024b910","requestLatencyInMs":223,"requestStartTimeUTC":"2024-04-30T16:36:27.452Z","requestEndTimeUTC":"2024-04-30T16:36:27.675Z","responseStatisticsList":[{"storeResult":{"storePhysicalAddress":"rntbd://cdb-ms-prod-australiacentral1-be12.documents.azure.com:14010/apps/4b626f38-121e-41de-8eb7-a0288e0c953e/services/84750cb0-bc12-4ef2-91b9-2355e93c3b91/partitions/3fae389f-7b54-4971-a5df-ab213506095c/replicas/133579598729731764s/","lsn":10520,"globalCommittedLsn":10519,"partitionKeyRangeId":"0","isValid":true,"statusCode":400,"subStatusCode":0,"isGone":false,"isNotFound":false,"isInvalidPartition":false,"isThroughputControlRequestRateTooLarge":false,"requestCharge":1.0,"itemLSN":-1,"sessionToken":"0:-1#10520","backendLatencyInMs":19.502,"retryAfterInMs":null,"exceptionMessage":"null","exceptionResponseHeaders":"{x-ms-last-state-change-utc=Fri, 19 Apr 2024 00:24:38.699 GMT, x-ms-request-duration-ms=19.502, x-ms-session-token=0:-1#10520, lsn=10520, x-ms-request-charge=1.00, x-ms-schemaversion=1.17, x-ms-transport-request-id=10, x-ms-number-of-read-regions=0, x-ms-resource-usage=documentSize=0;documentsSize=210;documentsCount=3;collectionSize=210;, x-ms-resource-quota=documentSize=51200;documentsSize=52428800;documentsCount=-1;collectionSize=52428800;, x-ms-activity-id=1ee27087-c5cb-4931-93e5-ce231024b910, x-ms-xp-role=2, x-ms-global-Committed-lsn=10519, x-ms-documentdb-partitionkeyrangeid=0, x-ms-cosmos-llsn=10520, x-ms-serviceversion= version=2.14.0.0}","replicaStatusList":["14010:Connected","14303:Connected","14011:Connected","14320:Connected"],"transportRequestTimeline":[{"eventName":"created","startTimeUTC":"2024-04-30T16:36:27.454Z","durationInMilliSecs":0.0},{"eventName":"queued","startTimeUTC":"2024-04-30T16:36:27.454Z","durationInMilliSecs":0.0},{"eventName":"channelAcquisitionStarted","startTimeUTC":"2024-04-30T16:36:27.454Z","durationInMilliSecs":1.0},{"eventName":"pipelined","startTimeUTC":"2024-04-30T16:36:27.455Z","durationInMilliSecs":2.0},{"eventName":"transitTime","startTimeUTC":"2024-04-30T16:36:27.457Z","durationInMilliSecs":217.0},{"eventName":"decodeTime","startTimeUTC":"2024-04-30T16:36:27.674Z","durationInMilliSecs":1.0},{"eventName":"received","startTimeUTC":"2024-04-30T16:36:27.675Z","durationInMilliSecs":0.0},{"eventName":"completed","startTimeUTC":"2024-04-30T16:36:27.675Z","durationInMilliSecs":0.0}],"rntbdRequestLengthInBytes":547,"rntbdResponseLengthInBytes":472,"requestPayloadLengthInBytes":89,"responsePayloadLengthInBytes":472,"channelStatistics":{"channelId":"c590194e","channelTaskQueueSize":0,"pendingRequestsCount":0,"lastReadTime":"2024-04-30T16:33:45.566Z","waitForConnectionInit":false},"serviceEndpointStatistics":{"availableChannels":1,"acquiredChannels":0,"executorTaskQueueSize":0,"inflightRequests":1,"lastSuccessfulRequestTime":"2024-04-30T16:26:53.159Z","lastRequestTime":"2024-04-30T16:33:45.346Z","createdTime":"2024-04-30T16:23:34.126Z","isClosed":false,"cerMetrics":{}}},"requestResponseTimeUTC":"2024-04-30T16:36:27.675Z","requestStartTimeUTC":"2024-04-30T16:36:27.454Z","requestResourceType":"Document","requestOperationType":"Query","requestSessionToken":null,"e2ePolicyCfg":null,"excludedRegions":null}],"supplementalResponseStatisticsList":[],"addressResolutionStatistics":{},"regionsContacted":["australia central"],"retryContext":{"statusAndSubStatusCodes":null,"retryLatency":0,"retryCount":0},"metadataDiagnosticsContext":{"metadataDiagnosticList":null},"serializationDiagnosticsContext":{"serializationDiagnosticsList":null},"gatewayStatisticsList":[],"samplingRateSnapshot":1.0,"systemInformation":{"usedMemory":"1224928 KB","availableMemory":"18348832 KB","systemCpuLoad":"(2024-04-30T16:35:58.925Z 10.8%), (2024-04-30T16:36:03.925Z 22.7%), (2024-04-30T16:36:08.925Z 10.2%), (2024-04-30T16:36:13.925Z 7.7%), (2024-04-30T16:36:18.925Z 15.1%), (2024-04-30T16:36:23.925Z 16.4%)","availableProcessors":4},"clientCfgs":{"id":1,"machineId":"uuid:32e35acb-c22a-4d70-b093-ddc8ef216862","connectionMode":"DIRECT","numberOfClients":1,"excrgns":"[]","clientEndpoints":{"https://cnx-cosmosdb-aus-dev.documents.azure.com:443/":2},"connCfg":{"rntbd":"(cto:PT10S, nrto:PT10S, icto:PT0S, ieto:PT1H, mcpe:130, mrpc:30, cer:true)","gw":"(cps:1000, nrto:PT1M, icto:PT1M, p:false)","other":"(ed: true, cs: false, rv: true)"},"consistencyCfg":"(consistency: Eventual, mm: true, prgns: [])","proactiveInitCfg":"","e2ePolicyCfg":"","sessionRetryCfg":""}}}
phemanth 6,475 Reputation points Microsoft Vendor

2024-05-01T04:27:07.8766667+00:00
@Andrew Hopkinson The error you're encountering when using a variable in your Cosmos DB query within Azure Data Factory (ADF) likely stems from the way you're referencing the variable. Here's a breakdown of the issue and the solution:

The Problem:

ADF uses a different syntax for referencing variables compared to directly embedding a value in the query.

Using $parameter1 is not the correct way to reference a variable within ADF's expression builder.

The Solution:

Access the Variable:

There are two common ways to access variables in ADF depending on where you defined the variable:

Pipeline Variable: Use the @pipeline().Variables['variableName'] syntax. For example, if your variable is named LastRunDateTS, access it with @pipeline().Variables['LastRunDateTS'].

Data Flow Variable: Use the @variables('variableName') syntax. This is similar to pipeline variables, but specific to the data flow context.

Update Your Query:

Modify your Cosmos DB query to use the appropriate syntax for referencing the variable:

SELECT * FROM c WHERE c._ts > @pipeline().Variables['LastRunDateTS'] -- For pipeline variable -- OR WHERE c._ts > @variables('LastRunDateTS') -- For data flow

Double-check the variable name to ensure it matches exactly with the one you defined in ADF.

Verify that the variable data type is compatible with the comparison in the query (e.g., string, integer).

If you're still encountering issues, consider using the ADF user interface to create and map variables. The UI might provide more guidance and prevent syntax errors.
phemanth 6,475 Reputation points Microsoft Vendor

2024-05-02T10:18:58.4466667+00:00

@Andrew Hopkinson We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others. Otherwise, will respond with more details and we will try to help.
Andrew Hopkinson 6 Reputation points

2024-05-02T15:45:36.1+00:00

This doesn't work, I get errors and it says that it is not a valid query expression. See my answer to see how I actually got it to work.

2 answers

Andrew Hopkinson 6 Reputation points

2024-05-02T15:54:53.0133333+00:00
This is what I had to do get it to work, the suggestions provided in this post do not work for me,First I created a variable the LastRunDateString

Then I created a parameter "parameter2" for the "Data Flow" in this value I put the following pipeline express:

@concat('SELECT * FROM c WHERE c._ts > ',variables('LastRunTimeString'))

This is a little wierd to get to work because there is also a parameter you need to add to your actual "dataflow" task (which is confusing). Then in the expression builder for the CosmosDB source SQL stateme, I put $parameter2 and it seems to work.
Please sign in to rate this answer.
phemanth 6,475 Reputation points Microsoft Vendor

2024-05-02T16:38:59.1833333+00:00

@Andrew Hopkinson Glad to know your issue has been resolved. Since the Microsoft Q&A community has a policy that "The question author cannot accept their own answer. They can only accept answers by others "I'll repost your solution in case you'd like to accept the answer.
Sign in to comment
phemanth 6,475 Reputation points Microsoft Vendor

2024-05-02T16:38:13.01+00:00
@Andrew Hopkinson

Welcome to Microsoft Q&A platform and thanks for posting your question.I'm glad that you were able to resolve your issue and thank you for posting your solution so that others experiencing the same thing can easily reference this! Since the Microsoft Q&A community has a policy that "The question author cannot accept their own answer. They can only accept answers by others "I'll repost your solution in case you'd like to accept the answer.

Ask: I am doing a simple data flow where I have a source from a cosmos db container, I want to get the latest data that is greater than a _ts value that I store in a variable (SELECT * FROM c WHERE c_ts > @variables('LastRunDateTS')). How can I pass this to query in the source for the cosmos query? I see there are parameters, but I don't want to hard code the values, I would like these to come from a variable and it gives errors when I try and reference a variable inside the query. Please help, thanks.

Solution: First I created a variable the LastRunDateString

Then I created a parameter "parameter2" for the "Data Flow" in this value I put the following pipeline express:

@concat('SELECT * FROM c WHERE c._ts > ',variables('LastRunTimeString'))

This is a little wierd to get to work because there is also a parameter you need to add to your actual "dataflow" task (which is confusing). Then in the expression builder for the CosmosDB source SQL stateme, I put $parameter2 and it seems to work.

If I missed anything please let me know and I'd be happy to add it to my answer, or feel free to comment below with any additional information.

If you have any other questions, please let me know. Thank you again for your time and patience throughout this issue.

Please don’t forget to Accept Answer and Yes for "was this answer helpful" wherever the information provided helps you, this can be beneficial to other community members.
Please sign in to rate this answer.
phemanth 6,475 Reputation points Microsoft Vendor

2024-05-03T09:20:15.4+00:00

@Andrew Hopkinson Following up to see if the above answer was helpful. If this answers your query, do click Accept Answer and Yes for was this answer helpful. And, if you have any further query do let us know.
Sign in to comment

Share via

How do I create a dynamic SQL query for Cosmos DB using Azure Data Factory?

2 answers