Führen Sie denselben Operator als mehrere Aufgaben in Cloud Composer aus

Frageninhalt

Ich habe einen Python-Operator im Luftstrom, der mit Cloud Composer ausgeführt wurde:

with DAG(
    dag_id = config['dag_id'],
    schedule_interval = config['schedule_interval'],
    default_args = default_args
    ) as dag:
    
    generate_data_task = PythonOperator(
        task_id = 'generate_dummy_data',
        python_callable = generate_data,
        dag = dag
    )

Die Funktion

generate_data() schreibt eine zufällig generierte, eindeutig benannte CSV-Datei in einen Bucket, der einige Daten enthält. Die Ausführung funktioniert einwandfrei, aber ich möchte dieselbe Aufgabe mehrmals parallel ausführen. Wenn ich 10 parallele Ausführungen spezifiziere, erwarte ich, dass 10 Dateien in den Bucket geschrieben werden. Ich habe Parallelität und task_concurrency ausprobiert, aber das gleiche Ergebnis erhalten.

Kann dies durch den Luftstrom auf dem Cloud Composer erreicht werden?

Richtige Antwort

Verwenden Sie Dynamische Aufgabenzuordnung:

generate_data_task = PythonOperator.partial(
        task_id = 'generate_dummy_data',
        python_callable = generate_data,
        dag = dag
    ).expand(op_args=[[]] * 10)

Das obige ist der detaillierte Inhalt vonFühren Sie denselben Operator als mehrere Aufgaben in Cloud Composer aus. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!