Einstellung bis s3 für die Protokolle in den Luftstrom

UPDATE Luftstrom 1.10 erleichtert die Protokollierung des viel einfacher ist.

Zur s3-logging stellen Sie die Verbindung Haken als pro die obige Antwort

und dann einfach fügen Sie den folgenden, um den Luftstrom.cfg

    [core]
    # Airflow can store logs remotely in AWS S3. Users must supply a remote
    # location URL (starting with either 's3://...') and an Airflow connection
    # id that provides access to the storage location.
    remote_base_log_folder = s3://my-bucket/path/to/logs
    remote_log_conn_id = MyS3Conn
    # Use server-side encryption for logs stored in S3
    encrypt_s3_logs = False

Für gcs Protokollierung,

1. Installieren Sie die gcp_api Paket zuerst, etwa so: pip install apache-Volumenstrom[gcp_api].

2. Einrichten der Verbindung Haken als pro die obige Antwort

3. Fügen Sie den folgenden, um den Luftstrom.cfg

   [core]
   # Airflow can store logs remotely in AWS S3. Users must supply a remote
   # location URL (starting with either 's3://...') and an Airflow connection
   # id that provides access to the storage location.
   remote_logging = True
   remote_base_log_folder = gs://my-bucket/path/to/logs
   remote_log_conn_id = MyGCSConn

HINWEIS: Da der Luftstrom 1.9 remote-logging wurde deutlich verändert. Wenn Sie mit 1.9, Lesen Sie weiter.

Referenz hier

Vollständige Anleitung:

1. Erstellen Sie ein Verzeichnis zum speichern von Konfigurationen und platzieren Sie dieses so, dass es sein kann gefunden in PYTHONPATH. Ein Beispiel ist $AIRFLOW_HOME/config

2. Erstellen leerer Dateien mit dem Namen $AIRFLOW_HOME/config/log_config.py und
   $AIRFLOW_HOME/config/__init__.py

3. Kopieren Sie den Inhalt des airflow/config_templates/airflow_local_settings.py in die log_config.py die Datei, die gerade erstellt wurde in dem oben genannten Schritt.

4. Anpassen der folgenden Teile von der Vorlage:

   #Add this variable to the top of the file. Note the trailing slash.
   S3_LOG_FOLDER = 's3://<bucket where logs should be persisted>/'
   
   Rename DEFAULT_LOGGING_CONFIG to LOGGING CONFIG
   LOGGING_CONFIG = ...
   
   Add a S3TaskHandler to the 'handlers' block of the LOGGING_CONFIG variable
   's3.task': {
       'class': 'airflow.utils.log.s3_task_handler.S3TaskHandler',
       'formatter': 'airflow.task',
       'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
       's3_log_folder': S3_LOG_FOLDER,
       'filename_template': FILENAME_TEMPLATE,
   },
   
    Update the airflow.task and airflow.task_runner blocks to be 's3.task' instead >of 'file.task'.
   'loggers': {
       'airflow.task': {
           'handlers': ['s3.task'],
           ...
       },
       'airflow.task_runner': {
           'handlers': ['s3.task'],
           ...
       },
       'airflow': {
           'handlers': ['console'],
           ...
       },
   }

5. Stellen Sie sicher, dass eine s3-Verbindung Haken definiert wurde, in den Luftstrom, als pro die obige Antwort. Die Haken haben sollte lese-und Schreibzugriff auf den s3-bucket oben definiert sind, in S3_LOG_FOLDER.

6. Update $AIRFLOW_HOME/Luftstrom.cfg enthalten:

   task_log_reader = s3.task
   logging_config_class = log_config.LOGGING_CONFIG
   remote_log_conn_id = <name of the s3 platform hook>

7. Starten Sie den Luftstrom webserver und der scheduler und einen trigger (oder warten) einen neuen task ausführen.

8. Stellen Sie sicher, dass die Protokolle zeigen, bis für die neu ausgeführten Aufgaben in die Eimer, die Sie definiert haben.

9. Stellen Sie sicher, dass die s3 storage viewer arbeitet in der Benutzeroberfläche. Ziehen Sie sich einen neu ausgeführt Aufgabe, und stellen Sie sicher, dass Sie etwas wie:

   *** Reading remote log from gs://<bucket where logs should be persisted>/example_bash_operator/run_this_last/2017-10-03T00:00:00/16.log.
   [2017-10-03 21:57:50,056] {cli.py:377} INFO - Running on host chrisr-00532
   [2017-10-03 21:57:50,093] {base_task_runner.py:115} INFO - Running: ['bash', '-c', u'airflow run example_bash_operator run_this_last 2017-10-03T00:00:00 --job_id 47 --raw -sd DAGS_FOLDER/example_dags/example_bash_operator.py']
   [2017-10-03 21:57:51,264] {base_task_runner.py:98} INFO - Subtask: [2017-10-03 21:57:51,263] {__init__.py:45} INFO - Using executor SequentialExecutor
   [2017-10-03 21:57:51,306] {base_task_runner.py:98} INFO - Subtask: [2017-10-03 21:57:51,306] {models.py:186} INFO - Filling up the DagBag from /airflow/dags/example_dags/example_bash_operator.py

Abgeschlossen Arne die Antwort mit dem letzten Luftstrom updates, die Sie nicht benötigen, um task_log_reader auf einen anderen Wert als den Standardwert ein : task

Als wenn Sie Folgen Sie den Standard-logging-Vorlage airflow/config_templates/airflow_local_settings.py können Sie sehen, seit diesem commit (Hinweis des Hundeführers name geändert's3': {'task'... statt s3.task) das ist der Wert, auf den remote-Ordner(REMOTE_BASE_LOG_FOLDER) ersetzen wird der handler mit der rechten:

REMOTE_LOGGING = conf.get('core', 'remote_logging')

if REMOTE_LOGGING and REMOTE_BASE_LOG_FOLDER.startswith('s3://'):
        DEFAULT_LOGGING_CONFIG['handlers'].update(REMOTE_HANDLERS['s3'])
elif REMOTE_LOGGING and REMOTE_BASE_LOG_FOLDER.startswith('gs://'):
        DEFAULT_LOGGING_CONFIG['handlers'].update(REMOTE_HANDLERS['gcs'])
elif REMOTE_LOGGING and REMOTE_BASE_LOG_FOLDER.startswith('wasb'):
        DEFAULT_LOGGING_CONFIG['handlers'].update(REMOTE_HANDLERS['wasb'])
elif REMOTE_LOGGING and ELASTICSEARCH_HOST:
        DEFAULT_LOGGING_CONFIG['handlers'].update(REMOTE_HANDLERS['elasticsearch'])

Mehr details auf, wie Sie sich zu/Lesen von S3 : https://github.com/apache/incubator-airflow/blob/master/docs/howto/write-logs.rst#writing-logs-to-amazon-s3

Habe es die Arbeit mit Luftstrom 10 in kube.
Ich habe Folgendes env var sets:

AIRFLOW_CONN_LOGS_S3=s3://id:secret_uri_encoded@S3
AIRFLOW__CORE__REMOTE_LOGGING=True
AIRFLOW__CORE__REMOTE_BASE_LOG_FOLDER=s3://xxxx/logs
AIRFLOW__CORE__REMOTE_LOG_CONN_ID=logs_s3