Es wird versucht, dynamische Unter-Tags aus dem übergeordneten Tag basierend auf einem Array von Dateinamen zu erstellen

10

Ich versuche, s3-Dateien mithilfe des Luftstroms aus einem "nicht löschenden" Bucket (dh ich kann die Dateien nicht löschen) in GCS zu verschieben. Ich kann nicht garantieren, dass jeden Tag neue Dateien vorhanden sind, aber ich muss jeden Tag nach neuen Dateien suchen.

Mein Problem ist die dynamische Erstellung von Subtags. Wenn es Dateien gibt, brauche ich Subtags. Wenn es KEINE Dateien gibt, brauche ich keine Subtags. Mein Problem sind die Upstream / Downstream-Einstellungen. In meinem Code erkennt es zwar Dateien, startet aber die Subtags nicht so, wie sie sollen. Ich vermisse etwas.

Hier ist mein Code:

from airflow import models
from  airflow.utils.helpers import chain
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.operators.python_operator import PythonOperator, BranchPythonOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.subdag_operator import SubDagOperator
from airflow.contrib.operators.s3_to_gcs_operator import S3ToGoogleCloudStorageOperator
from airflow.utils import dates
from airflow.models import Variable
import logging

args = {
    'owner': 'Airflow',
    'start_date': dates.days_ago(1),
    'email': ['[email protected]'],
    'email_on_failure': True,
    'email_on_success': True,
}

bucket = 'mybucket'
prefix = 'myprefix/'
LastBDEXDate = int(Variable.get("last_publish_date"))
maxdate = LastBDEXDate
files = []

parent_dag = models.DAG(
    dag_id='My_Ingestion',
    default_args=args,
    schedule_interval='@daily',
    catchup=False
)

def Check_For_Files(**kwargs):
    s3 = S3Hook(aws_conn_id='S3_BOX')
    s3.get_conn()
    bucket = bucket
    LastBDEXDate = int(Variable.get("last_publish_date"))
    maxdate = LastBDEXDate
    files = s3.list_keys(bucket_name=bucket, prefix='myprefix/file')
    for file in files:
        print(file)
        print(file.split("_")[-2])
        print(file.split("_")[-2][-8:])  ##proves I can see a date in the file name is ok.
        maxdate = maxdate if maxdate > int(file.split("_")[-2][-8:]) else int(file.split("_")[-2][-8:])
    if maxdate > LastBDEXDate:
        return 'Start_Process'
    return 'finished'

def create_subdag(dag_parent, dag_id_child_prefix, file_name):
    # dag params
    dag_id_child = '%s.%s' % (dag_parent.dag_id, dag_id_child_prefix)

    # dag
    subdag = models.DAG(dag_id=dag_id_child,
              default_args=args,
              schedule_interval=None)

    # operators
    s3_to_gcs_op = S3ToGoogleCloudStorageOperator(
        task_id=dag_id_child,
        bucket=bucket,
        prefix=file_name,
        dest_gcs_conn_id='GCP_Account',
        dest_gcs='gs://my_files/To_Process/',
        replace=False,
        gzip=True,
        dag=subdag)


    return subdag

def create_subdag_operator(dag_parent, filename, index):
    tid_subdag = 'file_{}'.format(index)
    subdag = create_subdag(dag_parent, tid_subdag, filename)
    sd_op = SubDagOperator(task_id=tid_subdag, dag=dag_parent, subdag=subdag)
    return sd_op

def create_subdag_operators(dag_parent, file_list):
    subdags = [create_subdag_operator(dag_parent, file, file_list.index(file)) for file in file_list]
    # chain subdag-operators together
    chain(*subdags)
    return subdags

check_for_files = BranchPythonOperator(
    task_id='Check_for_s3_Files',
    provide_context=True,
    python_callable=Check_For_Files,
    dag=parent_dag
)

finished = DummyOperator(
    task_id='finished',
    dag=parent_dag
)

decision_to_continue = DummyOperator(
    task_id='Start_Process',
    dag=parent_dag
)

if len(files) > 0:
    subdag_ops = create_subdag_operators(parent_dag, files)
    check_for_files >> decision_to_continue >> subdag_ops[0] >> subdag_ops[-1] >> finished


check_for_files >> finished
arcee123
quelle
Welche Art von Job im Backend dieser DAGS ausgeführt wird, sind diese sparkJobs oder ein pythonSkript und wie verwenden Sie sie, um sie auszuführen, livyoder eine andere Methode
ashwin agrawal
Es tut mir leid, ich verstehe die Frage nicht. Kannst du bitte noch einmal wiederholen?
arcee123
Ich meine, Sie verwenden nur einfache Python-Skripte und keinen Spark-Job, oder?
Ashwin Agrawal
Ja. einfache Bediener, die standardmäßig im Luftstrom sind. Ich möchte vorhandene Operatoren mit einer dynamischen Rate hinzufügen, die auf markierten Dateien in S3 basiert, die ich in GCS aufnehmen möchte.
arcee123
Warum ist fileseine leere Liste?
Oluwafemi Sule

Antworten:

3

Im Folgenden finden Sie die empfohlene Methode zum Erstellen einer dynamischen DAG oder Sub-DAG im Luftstrom. Es gibt jedoch auch andere Möglichkeiten, aber ich denke, dies ist weitgehend auf Ihr Problem anwendbar.

Erstellen Sie zunächst eine Datei, (yaml/csv)die die Liste aller s3Dateien und Speicherorte enthält. In Ihrem Fall haben Sie eine Funktion zum Speichern in einer Liste geschrieben. Ich würde sagen, speichern Sie sie in einer separaten yamlDatei und laden Sie sie zur Laufzeit in airflow env und erstellen Sie sie dann DAGs.

Unten ist eine Beispieldatei yaml: dynamicDagConfigFile.yaml

job: dynamic-dag
bucket_name: 'bucket-name'
prefix: 'bucket-prefix'
S3Files:
    - File1: 'S3Loc1'
    - File2: 'S3Loc2'
    - File3: 'S3Loc3'

Sie können Ihre Check_For_FilesFunktion ändern , um sie in einer yamlDatei zu speichern .

Jetzt können wir mit der dynamischen Dag-Erstellung fortfahren:

Definieren Sie zunächst zwei Aufgaben mit Dummy-Operatoren, die Start- und die Endaufgabe. Solche Aufgaben sind diejenigen, auf denen wir aufbauen werden, indem wir DAGdynamisch Aufgaben zwischen ihnen erstellen:

start = DummyOperator(
    task_id='start',
    dag=dag
)

end = DummyOperator(
    task_id='end',
    dag=dag)

Dynamische DAG: Wir werden PythonOperatorsim Luftstrom verwenden. Die Funktion sollte als Argumente die Task-ID erhalten. eine auszuführende Python-Funktion, dh python_callable für den Python-Operator; und eine Reihe von Argumenten, die während der Ausführung verwendet werden sollen.

Fügen Sie ein Argument hinzu, das task id. So können wir Daten zwischen Aufgaben austauschen, die auf dynamische Weise generiert wurden, z XCOM. B. über .

Sie können Ihre Operationsfunktion innerhalb dieses dynamischen Tages festlegen s3_to_gcs_op.

def createDynamicDAG(task_id, callableFunction, args):
    task = PythonOperator(
        task_id = task_id,
        provide_context=True,
        #Eval is used since the callableFunction var is of type string
        #while the python_callable argument for PythonOperators only receives objects of type callable not strings.
        python_callable = eval(callableFunction),
        op_kwargs = args,
        xcom_push = True,
        dag = dag,
    )
    return task

Basierend auf dem Speicherort in der Yaml-Datei können Sie schließlich dynamische Dags erstellen. Lesen Sie zuerst die yamlDatei wie folgt und erstellen Sie dynamische Dags :

with open('/usr/local/airflow/dags/config_files/dynamicDagConfigFile.yaml') as f:
    # use safe_load instead to load the YAML file
    configFile = yaml.safe_load(f)

    #Extract file list
    S3Files = configFile['S3Files']

    #In this loop tasks are created for each table defined in the YAML file
    for S3File in S3Files:
        for S3File, fieldName in S3File.items():

            #Remember task id is provided in order to exchange data among tasks generated in dynamic way.
            get_s3_files = createDynamicDAG('{}-getS3Data'.format(S3File), 
                                            'getS3Data', 
                                            {}) #your configs here.

            #Second step is upload S3 to GCS
            upload_s3_toGCS = createDynamicDAG('{}-uploadDataS3ToGCS'.format(S3File), 'uploadDataS3ToGCS', {'previous_task_id':'{}-'})

#write your configs again here like S3 bucket name prefix extra or read from yaml file, and other GCS config.

Endgültige DAG-Definition:

Die Idee ist das

#once tasks are generated they should linked with the
#dummy operators generated in the start and end tasks. 
start >> get_s3_files
get_s3_files >> upload_s3_toGCS
upload_s3_toGCS >> end

Vollständiger Luftstromcode in der Reihenfolge:

import yaml
import airflow
from airflow import DAG
from datetime import datetime, timedelta, time
from airflow.operators.python_operator import PythonOperator
from airflow.operators.dummy_operator import DummyOperator

start = DummyOperator(
    task_id='start',
    dag=dag
)


def createDynamicDAG(task_id, callableFunction, args):
    task = PythonOperator(
        task_id = task_id,
        provide_context=True,
        #Eval is used since the callableFunction var is of type string
        #while the python_callable argument for PythonOperators only receives objects of type callable not strings.
        python_callable = eval(callableFunction),
        op_kwargs = args,
        xcom_push = True,
        dag = dag,
    )
    return task


end = DummyOperator(
    task_id='end',
    dag=dag)



with open('/usr/local/airflow/dags/config_files/dynamicDagConfigFile.yaml') as f:
    configFile = yaml.safe_load(f)

    #Extract file list
    S3Files = configFile['S3Files']

    #In this loop tasks are created for each table defined in the YAML file
    for S3File in S3Files:
        for S3File, fieldName in S3File.items():

            #Remember task id is provided in order to exchange data among tasks generated in dynamic way.
            get_s3_files = createDynamicDAG('{}-getS3Data'.format(S3File), 
                                            'getS3Data', 
                                            {}) #your configs here.

            #Second step is upload S3 to GCS
            upload_s3_toGCS = createDynamicDAG('{}-uploadDataS3ToGCS'.format(S3File), 'uploadDataS3ToGCS', {'previous_task_id':'{}-'})

#write your configs again here like S3 bucket name prefix extra or read from yaml file, and other GCS config.


start >> get_s3_files
get_s3_files >> upload_s3_toGCS
upload_s3_toGCS >> end
Ashwin Agrawal
quelle
Ich danke dir sehr. Eines der Probleme, die ich hatte, war, was passiert, wenn es keine neuen Dateien gibt? Eines der Probleme, mit denen ich konfrontiert bin, ist, dass es immer Dateien an diesem Ort gibt, aber nicht garantiert wird, dass NEUE Dateien abgerufen werden, was bedeutet, dass der Abschnitt upload_s3_toGCSnicht existiert und Fehler im Luftstrom auftreten.
arcee123
Sie können das Problem lösen, indem Sie die Dateien aus der yamlDatei entfernen, sobald alle diese Dateien in GCS hochgeladen wurden. Auf diese Weise sind nur neue Dateien in der yamlDatei vorhanden. Und falls keine neuen Dateien vorhanden sind, ist die yamlDatei leer und es wird kein dynamischer Tag erstellt. Aus diesem Grund ist eine yamlDatei eine viel bessere Option als das Speichern von Dateien in einer Liste.
Ashwin Agrawal
Die yamlDatei hilft auch dabei, die Protokollierung der s3-Dateien auf eine Weise aufrechtzuerhalten. Wenn ein Teil der s3-Datei nicht in GCS hochgeladen werden kann, können Sie auch ein dieser Datei entsprechendes Flag verwalten und diese beim nächsten DAG-Lauf erneut versuchen.
Ashwin Agrawal
Und wenn es keine neuen Dateien gibt, können Sie ifder DAG eine Bedingung vorlegen, die nach neuen Dateien in yamlDateien sucht, wenn neue Dateien vorhanden sind. Führen Sie sie aus, andernfalls überspringen Sie sie.
Ashwin Agrawal
Das Problem hierbei ist, dass die Downstreams eingestellt sind. Wenn die Downstreams ohne die tatsächlichen Jobs festgelegt werden (da keine Dateien vorhanden sind), tritt ein Fehler auf.
arcee123