Создание кластера EMR с использованием запуска Airflow dag. После выполнения задачи EMR будет прервана.

Лучший способ сделать это, вероятно, иметь узел в корне вашей DAG Airflow, который создает кластер EMR, а затем другой узел в самом конце DAG, который раскручивает кластер после всех других узлов. завершено.

1180 Безусловно, это было бы наиболее эффективным использованием ресурсов. Позвольте мне предупредить вас: в этом много деталей; Я постараюсь перечислить столько, сколько вы получите. Я рекомендую вам добавить собственный исчерпывающий ответ, в котором перечислены все проблемы, с которыми вы столкнулись, и способ их обхода (как только вы это сделаете)

---

В отношении создания / прекращения работы кластера

• [1182 ] Для создания и завершения кластера у вас есть EmrCreateJobFlowOperator и EmrTerminateJobFlowOperator соответственно

• Не беспокойтесь, если вы не используете AWS SecretAccessKey (и полностью полагаться на IAM роли ); создание любого AWS -связанного hook или operator в Airflow будет автоматически отступать к базовой EC2 роли IAM

• Если вы НЕ используете EMR-Steps API для передачи задания, то вам также придется вручную определить обе вышеуказанные операции, используя Sensors ] . Уже есть датчик для фазы создания опроса, который называется EmrJobFlowSensor , и вы можете слегка изменить его, чтобы создать датчик для завершения

• Вы передаете свой кластер-конфигурацию JSON в job_flow_extra . Вы также можете передавать конфиги в параметре Connection (например, my_emr_conn) extra , но воздерживаться от него, потому что он часто прерывает загрузку SQLAlchemy ORM (так как это большая json)

---

Относительно отправки заданий

• Вы либо отправляете задания в Emr, используя EMR-Steps API , что можно сделать либо на этапе создания кластера (в JSON Cluster-Configs), либо впоследствии, используя add_job_flow_steps() . Есть даже emr_add_steps_operator() в Airflow, который также требует EmrStepSensor . Вы можете прочитать больше об этом в AWS документах , и вам, возможно, также придется использовать command-runner.jar

• Для конкретных случаев применения (например, Hive, Livy), вы можете использовать их конкретные способы. Например, вы можете использовать HiveServer2Hook , чтобы отправить Hive задание. Вот сложная часть: вызов run_job_flow() (сделанный на этапе создания кластера) дает вам только job_flow_id (идентификатор кластера). Вам нужно будет использовать вызов describe_cluster() , используя EmrHook , чтобы получить private-IP главного узла . Используя это, вы сможете программно создать Connection (например, Hive Server 2 Thrift соединение ) и использовать его для отправки ваших вычислений в кластер. И не забудьте удалить эти соединения (для удобства) до завершения рабочего процесса.

• Наконец, есть старый добрый bash для взаимодействия с кластером. Для этого вы также должны передать пару ключей EC2 на этапе создания кластера . После этого вы можете программно создать SSH соединение и использовать его (с SSHHook или SSHOperator ) для выполнения заданий в кластере. Узнайте больше о SSH-материалах в Airflow здесь

• Особенно для отправки Spark заданий на удаленных Emr кластера, прочитайте это обсуждение

---