如何从地图函数（Python）执行火花 sql 查询？答案

【问题标题】：How to execute a spark sql query from a map function (Python)?如何从地图函数（Python）执行火花 sql 查询？
【发布时间】：2016-05-16 00:56:19
【问题描述】：

如何从不是程序驱动部分的例程中执行 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 中注册的其他表。

【问题讨论】：

标签： pyspark apache-spark-sql parquet

【解决方案1】：

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

【讨论】：

【解决方案2】：

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

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

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

【讨论】：