Используйте оператор потока воздуха и вывод в текстовый файл

Question

Привет. Я хочу выполнить запрос улья с помощью оператора куста воздушного потока и вывести результат в файл. Я не хочу использовать INSERT OVERWRITE здесь.

hive_ex = HiveOperator(
    task_id='hive-ex',
    hql='/sql/hive-ex.sql',
    hiveconfs={
        'DAY': '{{ ds }}',
        'YESTERDAY': '{{ yesterday_ds }}',
        'OUTPUT': '{{ file_path }}'+'csv',
    },
    dag=dag
)

Каков наилучший способ сделать это?

Я знаю, как это сделать, используя оператор bash, но хочу знать, можем ли мы использовать оператор улья

hive_ex = BashOperator(
    task_id='hive-ex',
    bash_command='hive -f hive.sql -DAY={{ ds }} >> {{ file_path }} 
    /file_{{ds}}.json',
    dag=dag
)

Answer 1

Поскольку это довольно индивидуальный вариант использования, лучшим способом является расширение оператора Hive (или создание собственного Hive2CSVOperator). Реализация будет зависеть от того, есть ли у вас доступ к кустам через CLI или HiveServer2.

Улей CLI

Сначала я попытаюсь настроить соединение Hive CLI и добавить hive_cli_params согласно коду подключения Hive CLI , а если это не сработает, расширьте Hook (который даст вам доступ ко всему).

HiveServer2

Для этого случая есть отдельный хук ( ссылка ). Это немного удобнее, потому что у него есть метод get_results ( источник ) или to_csv метод ( источник ).

execute в коде оператора может выглядеть примерно так:

def execute():
  ...
  self.hook = HiveServer2Hook(...)
  self.conn = self.hook.get_conn()

  self.conn.to_csv(hql=self.hql, csv_filepath=self.output_filepath, ...)

Answer 2

вам нужны крючки воздушного потока. см. Крючки и HiveHook , есть метод to_csv или вы можете использовать метод get_records, а затем сделать это самостоятельно.