python - 来自 Hive 查询的持久 PySpark Dataframe

Question

我正在从 Hive 表中获取一些数据:

df = sqlContext.sql('select shubiru, date from thebigtable bt where bt.num > 10 ')
df.show() # here the query is processed and the results shown

而且一切正常。现在我想对 df 进行操作，但是每次我对 df 进行操作时，它都会再次运行针对 Hive 的查询:

import pyspark.sql.functions as func
from datetime import datetime
from pyspark.sql.types import TimestampType

dt_udt = func.udf(lambda x: datetime.strptime(str(x), '%Y%m%d') if x else None, TimestampType())
df = df.withColumn('fdate', dt_udt(df.date)) 
df.show()  # here the query is run again and the transformation is done

所以我认为如果我在 df 上调用 persist，查询将不会再次运行:

df.cache()
df = df.withColumn('fdate', dt_udf(df.date))

但没有骰子，查询再次针对 Hive 运行并由 UDF 处理。有没有一种方法可以在内存中缓存查询结果并在数据帧上运行操作而无需每次都访问 Hive？

Answer 1

只要对数据执行操作，Spark SQL 就会从 DataSource(在您的例子中是 Hive)中提取数据。在这种情况下，您试图在 cache() 之后重命名列，这将是无用的。我的建议是像

df = df.withColumn('fdate', dt_udf(df.date)).withColumn('date_column_2', dt_udf(df.date)).cache()

此语句之后的所有操作都将对持久保存在 spark 中的数据进行操作。然而，缓存大量数据会自动驱逐旧的 RDD 分区，并且需要返回到 hive 以重新生成丢失的分区。