Forum Discussion

RaymondLaw's avatar
RaymondLaw
Regular Visitor
3 years ago

End to End data pipeline across workspaces

Hi All,   Below is my planned architecture. It consists of two workspaces. They are in separated workspaces for security reason. (simplified from https://learn.microsoft.com/en-us/fabric/onelake/ge...
Need Help
pipelines
MartinMason's avatar
MartinMason
Icon for Resolver I rankResolver I
2 years ago

There's a workaround now that semantic-link has gone GA. With the workspaceId and pipelineId known, the Microsoft Fabric REST API can be used to invoke a data pipeline in a different workspace. But this will create the job scheduler instance of the pipeline. I've not figured out how to wait until the execution completes before returning as I can't get the freaking instanceId that should be returned from the call. Documentation on the Fabric API is poor.

import sempy.fabric as fabric

workspaceId = "<your workspace id>"
pipelineId = "<your pipeline id>"

client = fabric.FabricRestClient()
response = client.post(f"/v1/workspaces/{workspaceId}/items/{pipelineId}/jobs/instances?jobType=Pipeline")

 

MartinMason's avatar
MartinMason
Icon for Resolver I rankResolver I
2 years ago

Figured out how to get the instanceId from the Job Scheduler post back a while ago but forgot to put the code up here. Should anybody need to execute a pipeline in a different workspace synchronously, here's the code that works for me. Suggestions on how to improve this greatly welcome.

 

import sempy.fabric as fabric
import time

workspaceId = "{ workspaceId }"
pipelineId = "{ pipelineId }"
jobStatus = "NotStarted"
elapsedTime = 0

client = fabric.FabricRestClient()
response = client.post(f"/v1/workspaces/{workspaceId}/items/{pipelineId}/jobs/instances?jobType=Pipeline")
if response.status_code == 202:
    instanceId = response.headers["Location"][-36:]
    while (jobStatus != "Completed") & (jobStatus != "Cancelled") & (jobStatus != "Failed") & (elapsedTime < p_timeout_in_sec):
        time.sleep(10)        
        statusResponse = client.get(f"/v1/workspaces/{workspaceId}/items/{pipelineId}/jobs/instances/{instanceId}")
        jobStatus = statusResponse.json()["status"]
        elapsedTime += 10
mssparkutils.notebook.exit(f"Status of Pipeline Execution: {jobStatus}")