Конвейер обновления фабрики данных Python Azure

Question

Я хочу использовать Python для добавления действия в конвейер в фабрике данных Azure. С помощью следующего кода я заменяю фактическое действие, но не добавляю новое:

p_name = 'test'
act_name = 'Wait4'

Wait_activity = WaitActivity(name=act_name,wait_time_in_seconds=5)


p_obj = PipelineResource(activities=[Wait_activity])
p = adf_client.pipelines.create_or_update(rg_name, df_name, p_name, p_obj)

Это конвейер перед выполнением кода:

После запуска кода:

Ожидается:

Answer 1

Исследовал операторы в исходном коде:

Так что при обновлении конвейера свойство activities должно быть списком действий вконвейер, а не один.

Например:

wait_activity = WaitActivity(name="waittest", type="Wait", wait_time_in_seconds=100, )
ActivityDependency = [{"activity":"waittest","dependencyConditions":["Succeeded"]}]
wait_activity1 = WaitActivity(name="waittest1", type="Wait", wait_time_in_seconds=100,depends_on=ActivityDependency)


p_name = 'testforadf'
p_obj = PipelineResource(
        activities=[wait_activity, wait_activity1])
p = adf_client.pipelines.create_or_update(rg_name, df_name, p_name, p_obj)

Обратите внимание на две строки:

activities=[wait_activity, wait_activity1])

Это свойство должно содержать все ваши действия.

ActivityDependency = [{"activity":"waittest","dependencyConditions":["Succeeded"]}]

Это условия зависимости между вашими действиями.

Мой вывод:

Любая проблема, пожалуйста, сообщитея знаю.

---

Хорошо, пожалуйста, посмотрите мой пример кода:

Предпосылка состоит в том, что у меня уже есть два вышеупомянутых действия ожидания

adftest = adf_client.pipelines.get(rg_name,df_name,p_name)
print(adftest)
for activity in adftest.activities :
    print(activity.name)
    print(activity.type)

Тогда вывод будет:

{'additional_properties': None, 'id': '/subscriptions/b83c1ed3-c5b6-44fb-b5ba-2b83a074c23f/resourceGroups/v-jugong-ChinaCXPTeam/providers/Microsoft.DataFactory/factories/jaygongadf/pipelines/testforadf', 'name': 'testforadf', 'type': 'Microsoft.DataFactory/factories/pipelines', 'etag': 'ed006cf3-0000-0800-0000-5da970600000', 'description': None, 'activities': [<azure.mgmt.datafactory.models.wait_activity_py3.WaitActivity object at 0x000001C05FEDE0F0>, <azure.mgmt.datafactory.models.wait_activity_py3.WaitActivity object at 0x000001C05FED6DA0>], 'parameters': None, 'variables': None, 'concurrency': None, 'annotations': None, 'folder': None}
waittest
Wait
waittest1
Wait

Тогда вы сможете увидеть объекты в указанном выше свойстве activities. Кроме того, вы можете увидеть их типы: 'activities': [<azure.mgmt.datafactory.models.wait_activity_py3.WaitActivity object at 0x000001C05FEDE0F0>, <azure.mgmt.datafactory.models.wait_activity_py3.WaitActivity object at 0x000001C05FED6DA0>]

Они имеют тип WaitActivity, поэтому вы можете просмотреть их цикл активности, чтобы получить каждый элемент в нем, используя:

for activity in adftest.activities :
        print(activity.name)
        print(activity.type)

Вы можете просмотретькакие свойства содержит тип WaitActivity, например name, type в операторах исходного кода (для меня я использовал Pycharm для тестирования кода, среда IDE могла бы непосредственно определять исходный код)

Затем, если вы хотите добавить еще одно действие, например, еще одно WaitActivity:

wait_activity2 = WaitActivity(name="waittest2", type="Wait", wait_time_in_seconds=100, )
adftest.activities.append(wait_activity2)
p = adf_client.pipelines.create_or_update(rg_name, df_name, p_name, adftest)

Пожалуйста, смотрите код выше, я создал новый WaitActivity с именем wait_activity2затем добавьте его в массив activities. Затем обновите конвейер как обычно, вы найдете новое действие: