如何从映射函数 (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)

在此示例中，我尝试查询相同的 table，但也想查询在 Spark SQL 中注册的其他 table。

Answer 1

不对分布式数据执行嵌套操作structure.It Spark 根本不支持。您必须使用 joins、本地（可选广播）数据结构或直接访问外部数据。

Answer 2

如果您无法使用 joins 完成任务并希望运行内存中的 SQL 查询：

您可以考虑使用一些内存数据库，如 H2, Apache Derby 和 Redis 等，以更快地执行并行 SQL 查询，而不会失去内存计算的优势。

与 MySQL, PostgreSQL 等数据库相比，内存数据库将提供更快的访问速度。

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

How to execute a spark sql query from a map function (Python)?

parquet

apache-spark-sql

pyspark