如何将Dataframe类型的函数参数传递给SparkSQL查询

为了能够在SQL查询中访问DataFrame,请使用方法将其注册为临时视图 DataFrame.createOrReplaceTempView :

def my_function(df_table: DataFrame) -> DataFrame:
    temp_view_name = "some_temp_view"
    df_table.createOrReplaceTempView(temp_view_name)

    sql_query = f"""
    SELECT DISTINCT dt.CountryId,
    Cast(RIGHT(dt.RegionIdentifier, 2) as Integer) as RegionID
    FROM {temp_view_name} dt 
    WHERE dt.CountryCode = 23
    """

    df = spark.sql(sql_query)
    return df

您的SQL语句需要一个表的名称——您不能放DataFrame对象,它们不等效。DataFrame不一定有名称,因为它可能是您加载的DataFrame上的某些操作的结果。

如果您希望此函数始终接受表作为参数,那么只需传递名称(字符串)即可。

如果你预计有时它可能会得到一个DataFrame,这是其他操作的结果,那么你不能将select构建为SQL语句,但你可以通过应用pyspark函数来添加它,就像这样

 import pyspark.sql.functions as F

 df = df_table.where(F.col("CountryCode") == 23).\
               select(F.countDistinct(F.col("CountryId")), 
                      F.right(F.col("RegionIdentifier"),2).\
                                cast("integer").alias("RegionID"))