Как выполнить запрос искры sql из функции карты (Python)?

Question

Как выполнять искровые sql-запросы из подпрограмм, которые не являются частью драйвера программы?

from pyspark import SparkContext 
from pyspark.sql import SQLContext 
from pyspark.sql.types import * 


def doWork(rec): 
    data = SQLContext.sql("select * from zip_data where STATEFP ='{sfp}' and COUNTYFP = '{cfp}' ".format(sfp=rec[0], cfp=rec[1])) 
    for item in data.collect(): 
     print(item) 
    # do something 
    return (rec[0], rec[1]) 

if __name__ == "__main__": 
    sc = SparkContext(appName="Some app") 
    print("Starting some app") 

    SQLContext = SQLContext(sc) 

    parquetFile = SQLContext.read.parquet("/path/to/data/") 
    parquetFile.registerTempTable("zip_data") 


    df = SQLContext.sql("select distinct STATEFP,COUNTYFP from zip_data where STATEFP IN ('12') ") 
    rslts = df.map(doWork) 

    for rslt in rslts.collect(): 
     print(rslt) 

В этом примере я пытаюсь запросить ту же таблицу, но хотел бы запросить и другие таблицы, зарегистрированные в Spark SQL.

Answer 1

В распределенной структуре данных не выполняются вложенные операции. В Spark просто не поддерживается. Вы должны использовать joins, локальные (необязательно транслируемые) структуры данных или напрямую обращаться к внешним данным.

Answer 2

В случае, когда вы не можете выполнить свою задачу с joins и хотите запускать запросы SQL в памяти:

Вы можете рассмотреть возможность использования некоторой базы данных в оперативной памяти, как и H2, Apache DerbyRedis и т.д., чтобы выполнить параллельно быстрее SQL-запросы без потери преимуществ вычислений в памяти.

Базы данных в памяти обеспечат более быстрый доступ по сравнению с базами данных MySQL, PostgreSQL.