Как экспортировать результаты запроса SQL из блоков данных в хранилище озера данных Azure - PullRequest
Новая
       37

Как экспортировать результаты запроса SQL из блоков данных в хранилище озера данных Azure

0 голосов
/ 05 января 2019

Я пытаюсь экспортировать результаты запроса spark.sql в Databricks в папку в хранилище озера данных Azure - ADLS

Таблицы, к которым я обращаюсь, также находятся в ADLS.

Я получил доступ к файлам в ADLS из Databricks со следующим коммандой: 

base = spark.read.csv("adl://carlslake.azuredatalakestore.net/landing/",inferSchema=True,header=True)
base.createOrReplaceTempView('basetable')

Я запрашиваю таблицу с помощью следующей команды: 

try:
  dataframe = spark.sql("select * from basetable where LOAD_ID = 1199")
except:
  print("Exception occurred 1166")
else:
  print("Table Load_id 1166")

Затем я пытаюсь экспортировать результаты в папку в Azure, используя следующую команду: 

try:
 dataframe.coalesce(1).write.option("header","true").mode("overwrite").csv("adl://carlslake.azuredatalakestore.net/jfolder2/outputfiles/")
  rename_file("adl://carlslake.azuredatalakestore.net/jfolder2/outputfiles", "adl://carlslake.azuredatalakestore.net/landing/RAW", "csv", "Delta_LoyaltyAccount_merged")
except:
  print("Exception Occurred 1166")
else:
  print("Delta File Created")

Здесь есть две странные проблемы:

  1. Я указал запрос на load_id = 1199, и хотя load_id = 1199 не указан, запрос все еще успешен.

  2. Я бы хотел, чтобы второй оператор "try" потерпел неудачу, если первый оператор "try" завершился неудачно, но второй оператор try работает в отношении первого оператора "try".

Может кто-нибудь сообщить мне, где я иду не так?

Таблицу можно посмотреть здесь 1029 * таблица, с *

1 Ответ

0 голосов
/ 06 января 2019

Просто подумал, что поделюсь с вами ответом; 

try:
  dataframe = spark.sql("select * from basetable where LOAD_ID = 1166")
except:
  print("Exception occurred 1166")
if dataframe.count() == 0:
  print("No data rows 1166")
else:
  dataframe.coalesce(1).write.option("header","true").mode("overwrite").csv("adl://carlslake.azuredatalakestore.net/jfolder2/outputfiles/")
  rename_file("adl://carlslake.azuredatalakestore.net/jfolder2/outputfiles", "adl://carlslake.azuredatalakestore.net/landing/RAW", "csv", "Delta_LoyaltyAccount_merged")

Я надеюсь, что это работает и для вас.

...