Erstellen eines benutzerdefinierten Plugins mit Apache Hive und Hadoop

Amazon MWAA extrahiert den Inhalt eines plugins.zip bis/usr/local/airflow/plugins. Dies kann verwendet werden, um Binärdateien zu Ihren Containern hinzuzufügen. Darüber hinaus führt Apache Airflow beim Start den Inhalt der Python-Dateien im plugins Ordner aus, sodass Sie Umgebungsvariablen festlegen und ändern können. Das folgende Beispiel führt Sie durch die Schritte zur Erstellung eines benutzerdefinierten Plugins mit Apache Hive und Hadoop in einer Amazon Managed Workflows for Apache Airflow-Umgebung, das mit anderen benutzerdefinierten Plugins und Binärdateien kombiniert werden kann.

Version

• Der Beispielcode auf dieser Seite kann mit Apache Airflow v1 in Python 3.7 verwendet werden.

• Sie können das Codebeispiel auf dieser Seite mit Apache Airflow v2 in Python 3.10 verwenden.

Voraussetzungen

Um den Beispielcode auf dieser Seite zu verwenden, benötigen Sie Folgendes:

• Eine MWAAAmazon-Umgebung.

Berechtigungen

• Für die Verwendung des Codebeispiels auf dieser Seite sind keine zusätzlichen Berechtigungen erforderlich.

Voraussetzungen

Um den Beispielcode auf dieser Seite zu verwenden, fügen Sie Ihrem die folgenden Abhängigkeiten hinzurequirements.txt. Weitere Informationen hierzu finden Sie unter Python-Abhängigkeiten installieren.

Apache Airflow v2

-c https://raw.githubusercontent.com/apache/airflow/constraints-2.0.2/constraints-3.7.txt
apache-airflow-providers-amazon[apache.hive]

Apache Airflow v1

apache-airflow[hive]==1.10.12

Laden Sie Abhängigkeiten herunter

Amazon extrahiert MWAA den Inhalt von plugins.zip in /usr/local/airflow/plugins jeden MWAA Amazon-Scheduler- und Worker-Container. Dies wird verwendet, um Ihrer Umgebung Binärdateien hinzuzufügen. In den folgenden Schritten wird beschrieben, wie Sie die für das benutzerdefinierte Plugin benötigten Dateien zusammenstellen.

1. Navigieren Sie in der Befehlszeile zu dem Verzeichnis, in dem Sie Ihr Plugin erstellen möchten. Beispielsweise:

   cd plugins

2. Laden Sie Hadoop von einem Mirror herunter, zum Beispiel:

   wget https://downloads.apache.org/hadoop/common/hadoop-3.3.0/hadoop-3.3.0.tar.gz

3. Laden Sie Hive von einem Mirror herunter, zum Beispiel:

   wget https://downloads.apache.org/hive/hive-3.1.2/apache-hive-3.1.2-bin.tar.gz

4. Erstellen Sie ein Verzeichnis. Beispielsweise:

   mkdir hive_plugin

5. Extrahieren Sie Hadoop.

   tar -xvzf hadoop-3.3.0.tar.gz -C hive_plugin

6. Extrahieren Sie Hive.

   tar -xvzf apache-hive-3.1.2-bin.tar.gz -C hive_plugin

Benutzerdefiniertes Plugin

Apache Airflow führt beim Start den Inhalt der Python-Dateien im Plugins-Ordner aus. Dies wird verwendet, um Umgebungsvariablen festzulegen und zu ändern. In den folgenden Schritten wird der Beispielcode für das benutzerdefinierte Plugin beschrieben.

1. Navigieren Sie in der Befehlszeile zu dem hive_plugin Verzeichnis. Beispielsweise:

   cd hive_plugin

2. Kopieren Sie den Inhalt des folgenden Codebeispiels und speichern Sie ihn lokal wie hive_plugin.py im hive_plugin Verzeichnis.

   from airflow.plugins_manager import AirflowPlugin
   import os
   os.environ["JAVA_HOME"]="/usr/lib/jvm/jre"
   os.environ["HADOOP_HOME"]='/usr/local/airflow/plugins/hadoop-3.3.0'
   os.environ["HADOOP_CONF_DIR"]='/usr/local/airflow/plugins/hadoop-3.3.0/etc/hadoop'
   os.environ["HIVE_HOME"]='/usr/local/airflow/plugins/apache-hive-3.1.2-bin'
   os.environ["PATH"] = os.getenv("PATH") + ":/usr/local/airflow/plugins/hadoop-3.3.0:/usr/local/airflow/plugins/apache-hive-3.1.2-bin/bin:/usr/local/airflow/plugins/apache-hive-3.1.2-bin/lib" 
   os.environ["CLASSPATH"] = os.getenv("CLASSPATH") + ":/usr/local/airflow/plugins/apache-hive-3.1.2-bin/lib" 
   class EnvVarPlugin(AirflowPlugin):                
       name = 'hive_plugin' 
   

3. Kopieren Sie den Inhalt des folgenden Textes und speichern Sie ihn lokal wie .airflowignore im hive_plugin Verzeichnis.

   hadoop-3.3.0
   apache-hive-3.1.2-bin

Plugins.zip

Die folgenden Schritte zeigen, wie man erstelltplugins.zip. Der Inhalt dieses Beispiels kann mit anderen Plugins und Binärdateien in einer einzigen plugins.zip Datei kombiniert werden.

1. Navigieren Sie in der Befehlszeile zu dem hive_plugin Verzeichnis aus dem vorherigen Schritt. Beispielsweise:

   cd hive_plugin

2. Komprimieren Sie den Inhalt Ihres plugins Ordners.

   zip -r ../hive_plugin.zip ./

Codebeispiel

In den folgenden Schritten wird beschrieben, wie Sie den DAG Code erstellen, mit dem das benutzerdefinierte Plugin getestet wird.

1. Navigieren Sie in der Befehlszeile zu dem Verzeichnis, in dem Ihr DAG Code gespeichert ist. Beispielsweise:

   cd dags

2. Kopieren Sie den Inhalt des folgenden Codebeispiels und speichern Sie ihn lokal unterhive.py.

   from airflow import DAG
   from airflow.operators.bash_operator import BashOperator
   from airflow.utils.dates import days_ago
   
   with DAG(dag_id="hive_test_dag", schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
       hive_test = BashOperator(
           task_id="hive_test",
           bash_command='hive --help'
       )

Optionen für die Airflow-Konfiguration

Wenn Sie Apache Airflow v2 verwenden, fügen Sie es core.lazy_load_plugins : False als Apache Airflow-Konfigurationsoption hinzu. Weitere Informationen finden Sie unter Verwenden von Konfigurationsoptionen zum Laden von Plugins in 2.

Als nächstes

• Erfahren Sie unter, wie Sie die requirements.txt Datei in diesem Beispiel in Ihren Amazon S3 S3-Bucket hochladenPython-Abhängigkeiten installieren.

• Erfahren Sie unter, wie Sie den DAG Code in diesem Beispiel in den dags Ordner in Ihrem Amazon S3 S3-Bucket hochladenHinzufügen oder Aktualisieren von DAGs.

• Weitere Informationen zum Hochladen der plugins.zip Datei in diesem Beispiel in Ihren Amazon S3 S3-Bucket finden Sie unterInstallation benutzerdefinierter Plugins.