如何从map函数(Python)执行spark sql查询？

Question

如何从不是程序驱动程序部分的例程执行spark 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本地(可选广播)数据结构或直接访问外部数据.