Wir stellen vor: Pylint-Airflow

Ich freue mich, Ihnen die Veröffentlichung von Pylint-Airflow, einem Pylint-Plugin für Apache Airflow, vorstellen zu können.

Installieren Sie von PyPi aus:

pip install pylint-airflow

Und bewerben Sie sich mit:

pylint --load-plugins=pylint_airflow yourscript.py

Beispielhafte Ausgabe:

************* Modul airflow.contrib.example_dags.example_emr_job_flow_automatic_steps
airflow/contrib/example_dags/example_emr_job_flow_automatic_steps.py:1:0: C0111: Fehlendes Modul docstring  (fehlende Zeichenkette)
airflow/contrib/example_dags/example_emr_job_flow_automatic_steps.py:1:0: C8306: Aus Konsistenzgründen muss der DAG-Dateiname mit der dag_id übereinstimmen  (match-dagid-filename)
airflow/contrib/example_dags/example_emr_job_flow_automatic_steps.py:63:0: C8300: Operatorvariablenname und task_id Argument sollten übereinstimmen  (different-operator-varname-taskid)
airflow/contrib/example_dags/example_emr_job_flow_automatic_steps.py:71:0: C8300: Operatorvariablenname und task_id Argument sollten übereinstimmen  (different-operator-varname-taskid)

Pylint-Airflow wurde für das Linting (statische Codeanalyse) von Airflow-DAG-Skripten entwickelt. Es läuft zusätzlich zu Ihren Standard-Pylint-Einstellungen und benachrichtigt Sie über mögliche Konventionen, Refactorings, Warnungen und Fehler. Das Plugin läuft mit Python 3.6 und höher. Sie können das Plugin in Ihre CI-Pipeline integrieren, um nach potenziellen Problemen zu suchen, und es ist sowohl über .pylintrc als auch inline wie andere Pylint-Meldungen konfigurierbar.

Das Plugin ist das Ergebnis eines unserer GDD-Freitage (plus etwas Extrazeit), einem 4-wöchentlichen Freitag, an dem wir einen ganzen Tag zum Experimentieren haben.

Beispiele

Auf der Grundlage meiner Erfahrungen mit Projekten und Open-Source-Beiträgen habe ich mir einige Konventionen ausgedacht, die ich für alle DAG-Skripte als nützlich erachte. Zum Beispiel:

Gemischte Abhängigkeitsrichtungen

Diese Meldung wird ausgegeben, wenn der Bitshift-Operator zum Setzen von Task-Abhängigkeiten in entgegengesetzter Richtung auf einer einzigen Zeile angewendet wird:

t1 = DummyOperator(aufgabe_id="Aufgabe1")
t2 = DummyOperator(aufgabe_id="Aufgabe2")
t3 = DummyOperator(aufgabe_id="Aufgabe3")

t1 >> t3  t2

$ pylint --load-plugins=pylint_airflow beispiel.py
************* Beispiel für ein Modul
example.py:5:9: C8302: Vermeiden Sie die Vermischung der Richtungen der Aufgabenabhängigkeit  (mixed-dependency-directions)

Die Vermischung von Abhängigkeitsrichtungen wird im Allgemeinen als verwirrend angesehen. Eine bessere Alternative wäre es, die Richtung zu wechseln oder die Aussage auf mehrere Zeilen aufzuteilen:

[t1, t2] >> t3

# Oder:
t1 >> t3
t2 >> t3

Stimmen Sie die task_id mit dem Variablennamen ab

Aus Gründen der Lesbarkeit möchte ich die task_id eines Operators mit dem Namen der Variablen, der er zugewiesen ist, abgleichen:

# Richtig
tun_etwas = DummyOperator(aufgabe_id="etwas_tun")

# Falsch
tun_etwas_anderes = DummyOperator(aufgabe_id="zufälliger_name")

Die zweite Zuweisung ist gültiger Python-Code, aber das Plugin wird eine Konventionsmeldung zurückgeben:

$ pylint --load-plugins=pylint_airflow beispiel.py
************* Beispiel für ein Modul
example.py:5:0: C8300: Operator-Variablenname und task_id Argument sollten übereinstimmen  (different-operator-varname-taskid)

Warnung für unbenutzte XComs

Vielleicht eine magische "Funktion" von Airflow; ein Rückgabewert aus der Callable des PythonOperators wird automatisch als XCom im Airflow-Metaspeicher gespeichert. Wenn die Rückgabeanweisung unbeabsichtigt ist und das XCom nie verwendet (d.h. gezogen) wird, füllt sich der Metaspeicher mit veralteten Daten, auf die nie zugegriffen wird. Abgesehen von den überflüssigen Werten könnte dies zu einem Problem werden, wenn die zurückgegebenen Daten sehr groß sind.

Um diese Situation zu vermeiden, prüft das Plugin, ob xcom_pull() mit task_ids=[task_id of pushtask]im Skript enthalten ist, so dass es weiß, ob der XCom-Schlüssel gelesen wird und somit beabsichtigt ist oder nicht. Wenn jedoch kein Verweis auf den XCom-Schlüssel gefunden werden kann, wird eine Meldung zurückgegeben, um auf das möglicherweise nicht verwendete XCom hinzuweisen (es kann natürlich nicht wissen, ob es in anderen Skripten verwendet wird).

von airflow.operators.python_operator importieren PythonOperator

def _Pushtask():
    drucken("Dinge tun")
    return "foobar"  # Wert wird hier zurückgegeben, aber nicht anderweitig verwendet

pushtask = PythonOperator(aufgabe_id="pushtask", python_aufrufbar=_Pushtask)

def _Zielaufgabe():
    drucken("etwas tun")

Pulltask = PythonOperator(aufgabe_id="Pulltask", python_aufrufbar=_Zielaufgabe)

# Eine gültige Pulltask wäre:

# def _pulltask(task_instance, **_):
# print(task_instance.xcom_pull(task_ids="pushtask"))

# pulltask = PythonOperator(task_id="pulltask", python_callable=_pulltask, provide_context=True)

Der Wert "foobar" wird von _pushtask zurückgegeben, aber der XCom-Schlüssel, in dem er gespeichert ist, wird nirgendwo anders im Skript verwendet. Daher wird eine Refactoring-Meldung angezeigt:

$ pylint --load-plugins=pylint_airflow beispiel.py
************* Beispiel für ein Modul
example.py:7:11: R8300: Rückgabewert von _pushtask wird als XCom gespeichert, aber nirgends verwendet  (unbenutzt-xcom)

Die Zukunft

Pylint-Airflow befindet sich derzeit in der Alpha-Phase und erfordert noch weitere Arbeit. Im jetzigen Zustand habe ich versucht, mich hauptsächlich auf die Stabilität zu konzentrieren, damit es nirgendwo abstürzt. Es gibt jedoch noch viele Dinge zu erledigen:

• Machen Sie es besser konfigurierbar.
• Hinzufügen weiterer Meldungen. Einige Meldungen in der Readme sind derzeit nicht implementiert.
• Bessere Nachrichtentexte, z.B. different-operator-varname-taskid könnten auch den tatsächlichen Variablennamen und die task_id zurückgeben.
• Dokumentation schreiben. Die aktuelle Dokumentation wird automatisch erstellt und auf Read the Docs gehostet, ist aber leer.

Tests und Beiträge sind sehr willkommen. Wenn Sie auf Probleme stoßen oder einen Beitrag zum Projekt leisten möchten, erstellen Sie bitte ein Issue oder einen Pull Request auf GitHub: Pylint-Airflow.

Sie können mich gerne über alles, was mit Airflow zu tun hat, auf dem Airflow Slack @BasPH kontaktieren.

Übrigens, wir haben einen großartigen Apache Airflow-Kurs, in dem Sie die Interna, die Terminologie und die besten Praktiken für die Arbeit mit Airflow lernen können. bietet Ihnen praktische Erfahrungen beim Schreiben und Verwalten von Datenpipelines.