Search code examples
azure-data-factorypipelinekqlazure-data-explorer

How to capture output query from ADF pipeline from For Each activity

I have a pipeline on ADF with a For Loop activity and inside I have a Data Flow running a Kusto query. In this loop I am using a variable with 10 string in an array, so that my Data Flow will run based on those values as a parameter.

enter image description here

What is happening is, some of the runs in the loop is failing, and some others is working fine.

Example as image below:

enter image description here

How can I see the query that is running inside the Data Flow with all parameters and expression that I am using? I mean, the final query as an output. I would like to run that same query on my Kusto to see if something is wrong or not.

Based on @Rakesh Govindula answer, this is my data flow and this what I have tried from his answer:

enter image description here

Solution

  • Dataflow activity won't give the input query in its input or output. As a workaround try the below method.

    Take a branch to your dataflow and use derived column transformation. In this create a column and give the same query expression to this.

    You can add this by taking a branch to your current source or you can create another sample source and continue the same approach.

    enter image description here

    Here, I have taken SQL query for sample.

    It will build the query and stores the query string in the column query. Use cache sink to know this query at pipeline level. The write order for this sink should be 1.

    enter image description here

    In the cache sink mapping, filter the query column using Rule based mapping like below.

    enter image description here

    In the Dataflow activity, set the logging to None and select first row only.

    enter image description here

    Now, you can find the input query in the Dataflow output cache sink.

    enter image description here

    Use the below expression to use this in the pipeline.

    @activity('Data flow1').output.runStatus.output.sinkCache.value[0].query