Scala & Spark: Recykling instrukcji SQL

Question

Spędziłem sporo czasu, aby zakodować wiele zapytań SQL, które były wcześniej używane do pobierania danych dla różnych skryptów R. Jak to działa

sqlContent = readSQLFile("file1.sql") 
sqlContent = setSQLVariables(sqlContent, variables) 
results = executeSQL(sqlContent) 

Kluczem jest to, że dla niektórych zapytań wynika z uprzedniego zapytania jest wymagane - dlaczego tworzenie VIEW s w bazie sama nie rozwiązuje tego problemu. Z Spark 2.0 już zorientowali się sposób to zrobić tylko, że przez

// create a dataframe using a jdbc connection to the database 
val tableDf = spark.read.jdbc(...) 
var tempTableName = "TEMP_TABLE" + java.util.UUID.randomUUID.toString.replace("-", "").toUpperCase 
var sqlQuery = Source.fromURL(getClass.getResource("/sql/" + sqlFileName)).mkString 
sqlQuery = setSQLVariables(sqlQuery, sqlVariables) 
sqlQuery = sqlQuery.replace("OLD_TABLE_NAME",tempTableName) 
tableDf.createOrReplaceTempView(tempTableName) 
var data = spark.sql(sqlQuery) 

Ale to moim skromnym zdaniem bardzo skomplikowanego. Również bardziej złożone zapytania, np. zapytania, które obecnie nie działają w przypadku incooporate subcompery. Czy jest bardziej solidny sposób jak re-implementacji kodu SQL w kod Spark.SQL użyciu filter($""), .select($"") itp

Ogólnym celem jest uzyskanie wielu org.apache.spark.sql.DataFrame S, każda reprezentująca wyniki jednego zapytania byłego SQL (który zawsze kilka: JOIN s, WITH s itp.). Tak więc zapytania o numer n prowadziły do ​​nDataFrame.

Czy istnieje lepsza opcja niż podane dwa?

Setup: Hadoop v.2.7.3, Spark 2.0.0, Intelli J IDEA 2016.2, Scala 2.11.8, Testcluster na Win7 Workstation

Answer 1

To nie jest szczególnie jasne, co się wymogiem jest, ale myślę, że mówisz masz zapytaniami coś takiego:

SELECT * FROM people LEFT OUTER JOIN places ON ... 
SELECT * FROM (SELECT * FROM people LEFT OUTER JOIN places ON ...) WHERE age>20 

i co chcesz zadeklarować i wykonać to sprawnie jak

SELECT * FROM people LEFT OUTER JOIN places ON ... 
SELECT * FROM <cachedresult> WHERE age>20 

Aby to osiągnąć, poprawiłbym plik wejściowy, aby każda instrukcja SQL miała przypisaną nazwę tabeli, w której będzie przechowywany wynik.

np.

PEOPLEPLACES\tSELECT * FROM people LEFT OUTER JOIN places ON ... 
ADULTS=SELECT * FROM PEOPLEPLACES WHERE age>18 

Następnie wykonać w pętli jak

parseSqlFile().foreach({case (name, query) => { 
    val data: DataFrame = execute(query) 
    data.createOrReplaceTempView(name) 
} 

Upewnij się zadeklarować zapytań w kolejności, więc wszystkie wymagane tabele zostały utworzone. Inne robią trochę więcej analizowania i sortowania według zależności.

W RDMS nazwałbym te zmaterializowane widoki. to jest przekształcenie innych danych, takich jak widok, ale z wynikiem buforowanym do późniejszego ponownego wykorzystania.