如何从地图函数(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.

推荐答案

One 不对分布式数据结构执行嵌套操作.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屋！