Luftstrom bekommen Ergebnis nach der Ausführung eines operators

Habe ich konfiguriert Luftstrom erstellt und einige Dags und subDags, die rufen mehrere Operatoren.

Mein Problem ist, dass, wenn ein Betreiber läuft und beendet den job, ich würde gerne die Ergebnisse wieder in einigen python-Struktur.
Zum Beispiel:

File1.py

  ...
    ...
    sub_dag_one=SubDagOperator(subdag=subdag_accessHive(
PARENT_DAG_NAME, CHILD_DAG_NAME, default_args, STEP, macros,path,
       ),
        task_id=DELP_DAG_NAME,
        dag=dag,
    )

File2.py

  from airflow import DAG
    from airflow.operators import HiveOperator
def subdag_callHive(parent, child, args, step,
                         user_defined_macros, path
                        ):
        dag_subdag = DAG(
            dag_id='%s.%s' % (parent, child),
            default_args=args,
            schedule_interval="@daily",
            template_searchpath=path,
            user_defined_macros=user_defined_macros,
        )

        # some work...

        HiveOperator(
            task_id='some_id',
            hiveconf_jinja_translate=True,
            hql='select field1 from public.mytable limit 4;',
            trigger_rule='all_done',
            dag=dag_subdag,
        )

        return dag_subdag

Die Funktion subdag_callHive aufgerufen wird von einem anderen python-script, wo die wichtigsten Dag definiert ist, und die anderen benötigten Parameter.

Nur muss ich möchte in der Lage sein, um das Ergebnis aus der HiveOperator
(*select * from public.mytable Grenzwert 4;*), die 4 Werte in diesem Fall.

den zurückgegebenen dag_subdag ist ein Objekt < Klasse 'Luftstrom.Modelle.DAG' > und enthält alle Attribute/Daten an den Anruf, aber keine Informationen darüber, was die HiveOperator haben.

Ist das möglich? wenn ja, wie kann es erreicht werden.

InformationsquelleAutor Alg_D | 2016-06-26
  1. 7

    Hooks können Sie für Ihren Bedarf. Grundsätzlich ist die HiveOperator das gleiche tut, ruft er die Hive-Haken, die mehrere Methoden zur Arbeit mit den Ergebnissen.

    Verwenden PythonOperator eine Funktion aufzurufen, die dann beginnt ein Bienenstock Haken.

    Folgenden Beispiel könnte Ihnen dabei helfen.

    Code-Snippet:

    callHook = PythonOperator(
    task_id='foo',
    python_callable=do_work,
    dag=dag
)

def do_work():
    hiveserver = HiveServer2Hook()
    hql = "SELECT COUNT(*) FROM foo.bar"
    row_count = hiveserver.get_records(hql, schema='foo')
    print row_count[0][0]

    Alle verfügbaren Methoden finden Sie hier: https://github.com/apache/incubator-airflow/blob/master/airflow/hooks/hive_hooks.py

    InformationsquelleAutor Thees
Schreibe einen Kommentar