Если вам нужна дополнительная информация об использовании облачных функций для триггеров DAG на основе событий, я рекомендую вам прочитать эту документацию . Здесь объясняются все шаги по настройке облачной функции.
Читая инструкцию по ссылке выше, есть топи c «Создание вашей функции» , там вы можете увидеть пример как создать файлы main.py и Requeriment.txt . Кроме того, вы также можете увидеть эту ссылку для более подробного объяснения о написании облачной функции.
На GitHub также есть примеры кодов для Composer, которые могут быть полезно, чтобы вы начали.
Наконец, я хотел бы обратить ваше внимание, установили ли вы зависимости python и подключили Cloud Composer к другим средам, см. эту ссылку для более подробной информации.
ОБНОВЛЕНИЕ:
Относится к передаче параметров из облачной функции в DAG, это зависит от того, какие параметры вы хотите передать как переменную. В соответствии с этой документацией , вы можете передавать переменные как дату выполнения или имя файла как переменные. Для имени файла это может выглядеть следующим образом:
input_file = '{{dag_run.conf ["name"]}}'
Тогда в пределах ваш BashOperator установил «bash_command = 'echo' + input_file)»
Настройте поле «имя» dag_run.conf ⇒ gcloud beta composer среды запускают myenvironment --location = europe- west1 trigger_dag - - c '{"name": "Hello World"}' myDAG
Я рекомендую вам прочитать о Jinja Template и также PythonOperator (также это для PythonOperator). Кроме того, существует context dict , который помогает получить доступ к переменным из Airflow.