如何从地图函数(Python)执行Spark SQL查询? [英] How to execute a spark sql query from a map function (Python)?

问题描述

如何从不是程序驱动程序部分的例程中执行spark sql查询?

How does one execute spark sql queries from routines that are not the driver portion of the program?

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中注册的其他表.

In this example I'm attempting to query the same table but would like to query other tables registered in Spark SQL too.

推荐答案

一个不对分布式数据结构执行嵌套操作.Spark根本不支持它.您必须使用joins，本地(可选广播)数据结构或直接访问外部数据.

One does not execute nested operations on distributed data structure.It is simply not supported in Spark. You have to use joins, local (optionally broadcasted) data structures or access external data directly instead.

这篇关于如何从地图函数(Python)执行Spark SQL查询?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！