Запуск конвейера фабрики данных Azure из приложения логики с параметром

Question

Позвольте мне предвосхитить, что я новичок в Logic Apps and Data Factory. В любом случае, в настоящее время я работаю над интеграцией, и одной из ее частей является необходимость запуска конвейера в фабрике данных из приложений логики. Я успешно сделал это, часть one , которую я не могу понять, это как передать параметры в мой конвейер. Я пытался изменить JSON в разделах «параметры» и «триггеры», но пока не получил ничего, чтобы щелкнуть. Конвейер завершается выполнением, но только с параметрами по умолчанию.

Кто-нибудь уже достиг успеха в этом? Любая помощь приветствуется.

Answer 1

Я использовал решение DraganB, но подпись звонка на

CreateRunResponse runResponse = client.Pipelines.CreateRun(resourceGroup, dataFactoryName, "pipeline1", parameters);

изменилось. Незначительные правки делают эту работу отлично:

CreateRunResponse runResponse = client.Pipelines.CreateRun(resourceGroup, dataFactoryName, "pipeline1", parameters: parameters);

Вот функция для тех, кому это нужно.

[FunctionName("DatafactoryShim")]
    public async static Task<HttpResponseMessage> Run(
        [HttpTrigger(AuthorizationLevel.Function, "post")]
        HttpRequestMessage req,
        ExecutionContext context,
        TraceWriter log
    )
    {
        string messageBody = await req.Content.ReadAsStringAsync();

        BlobToDatalakeFactoryParameters postValues = JsonHelper.ToClass<BlobToDatalakeFactoryParameters>(messageBody);

        Dictionary<string, object> parameters = new Dictionary<string, object>
        {
            {"blobContainer", postValues.BlobContainer},
            {"blobFolder", postValues.BlobFolder },
            {"relativeDatalakeFolder", postValues.RelativeDatalakeFolder },
            {"modelType", postValues.ModelType }

        }; //this is how you add initialaze parameters

        var client = Authenticate(); //Authentication with azure

        string resourceGroup = ConfigurationManager.AppSettings["resourceGroup"];
        string dataFactoryName = ConfigurationManager.AppSettings["dataFactoryName"];
        string pipelineName = ConfigurationManager.AppSettings["pipelineName"];

        Console.WriteLine("Creating pipeline run...");
        CreateRunResponse runResponse = client.Pipelines.CreateRunWithHttpMessagesAsync(
            resourceGroup,
            dataFactoryName,
            pipelineName,
            parameters: parameters).Result.Body;
        Console.WriteLine("Pipeline run ID: " + runResponse.RunId);

        var response = new HttpResponseMessage();

        if (client.PipelineRuns.Get(ConfigurationManager.AppSettings["resourceGroup"],
            ConfigurationManager.AppSettings["dataFactoryName"], runResponse.RunId).Status.Equals("InProgress"))
        {
            response = new HttpResponseMessage(HttpStatusCode.OK)
            {
                Content = new StringContent(runResponse.RunId, Encoding.UTF8)
            };
        }
        else
        {
            response = new HttpResponseMessage(HttpStatusCode.BadRequest)
            {
                Content =
                    new StringContent("Pipeline didn't started", Encoding.UTF8) //just some validation for function
            };
        }

        return response;
    }

Answer 2

Вы можете использовать свойство body действия логического приложения «Создать конвейер» для передачи параметров в конвейер. Как всегда, будьте осторожны, потому что это действие не только в режиме предварительного просмотра, но я также не смог найти это решение ни в одной документации MS. Я только что сделал обоснованное предположение, основанное на том, как другие подобные действия отформатированы.

Пример:

"Run_my_pipeline": {
  "inputs": {
    "host": {
      "connection": {
        "name": "@parameters('$connections')['azuredatafactory']['connectionId']"
      }
    },
    "method": "post",
    "body": {
      "param1": "myParamValue",
      "param2": "myParamValue"
    },
    "path": "...",
    "queries": {
      "x-ms-api-version": "2017-09-01-preview"
    },
    "authentication": "@parameters('$authentication')"
  }
}

Answer 3

, как я уже сказал в комментарии, я создаю обходной путь с функциями лазури. Функции Azure и приложение логики хорошо работают вместе. По этой ссылке вы можете увидеть, как создавать и управлять конвейерами с помощью .net https://docs.microsoft.com/en-us/azure/data-factory/quickstart-create-data-factory-dot-net

Если у вас уже есть ADF и конвейер, вы просто хотите запустить его (с конвейерами), тогда вы можете просто

Dictionary<string, object> parameters = new Dictionary<string, object>
       {
           {"BoxSerialNumbers", req.BoxSerialNumbers},
           {"StartDate", req.StartDate },
           {"EndDate",req.EndDate },
           {"Recipient", req.Recipient }
       };//this is how you add initialaze parameters

        var client = Authenticate(); //Authentication with azure
        log.Info("Creating.");
        CreateRunResponse runResponse = client.Pipelines.CreateRun(resourceGroup, dataFactoryName, "pipeline1", parameters);//run pipeline, you can do this async (it's better)
        log.Info("Created.");
        var response = new HttpResponseMessage();
        if (client.PipelineRuns.Get(resourceGroup, dataFactoryName, runResponse.RunId).Status.Equals("InProgress"))
        {
            response = new HttpResponseMessage(HttpStatusCode.OK)
            {
                Content = new StringContent(runResponse.RunId, Encoding.UTF8)
            };
        }
        else
        {
            response = new HttpResponseMessage(HttpStatusCode.BadRequest)
            {
                Content = new StringContent("Pipeline didn't started", Encoding.UTF8)//just some validation for function
            };
        }
        return response;                                               


    public static DataFactoryManagementClient Authenticate()
    {
        var context = new AuthenticationContext("https://login.windows.net/" + tenantID);
        ClientCredential cc = new ClientCredential(applicationID, authenticationKey);
        AuthenticationResult result = context.AcquireTokenAsync("https://management.azure.com/", cc).Result;
        ServiceClientCredentials cred = new TokenCredentials(result.AccessToken);
        return new DataFactoryManagementClient(cred) { SubscriptionId = subscriptionID };
    }

Таким образом, в запросе вы можете передать свои параметры из приложения логики, с помощью runId вы можете проверить статус. Тогда в логике приложения просто простой HTTP-запрос для вызова этой функции. Надеюсь, это кому-нибудь поможет.