【问题标题】:Using Spark Data Frame or GlobalTempView with PySpark将 Spark Data Frame 或 GlobalTempView 与 PySpark 一起使用
【发布时间】:2020-05-23 01:59:44
【问题描述】:

我对 Spark 很陌生。 我设法将一个 PySpark 数据帧(在 %python 上下文中创建)传递给一个 Spark 数据帧(用于 Scala 单元)。但我需要知道的是如何做相反的事情:使用在 Scala 单元中的 %python 上下文中创建的数据框。

这里是详细信息。我已经阅读了 txt 文件中的两个表并将它们传递给 PySpark Dataframes。我还将它们传递给 GlobalTempViews。

%python

#Pass txt curated files to Data Frames
df_a = spark.read.option("header","true").option("delimiter","|").csv(curated_path_a) 
df_b = spark.read.option("header","true").option("delimiter","|").csv(curated_path_b) 

df_a.createOrReplaceTempView("df_a")
df_b.createOrReplaceTempView("df_b")

然后我需要使用 scala 代码,所以我将 PySpark 数据帧传递给 spark 表

//Passing Pyspark Data Frames to Spark Data Frames
val df_a = spark.table("df_a")
val df_b = spark.table("df_b")

然后我加入了这些表并获得了另一个名为 df_joined_tables 的数据框。我将它传递给 GlobalTempView。

//Creating scala variables for each table
var table_name_a = dbutils.widgets.get("table_name_a")
var table_name_b = dbutils.widgets.get("table_name_b")

//Create scala variable for Join ID
var join_id = dbutils.widgets.get("table_name_b") + "ID"
//var join_id = table_name_b + "ID"

// Define join type
var join_type = dbutils.widgets.get("join_type")

// Joining Tables
var df_joined_tables = df_a
                                     .join(df_b,                                               
                                               df_a(join_id)===df_b(join_id),
                                              join_type
                                          ).select($"df_a.*",$"df_b.Description".alias(table_name_b))

df_joined_tables.createOrReplaceGlobalTempView("df_joined_tables")
display(df_joined_tables)

现在我需要在 %python 上下文中使用这个 df_joined_tables。我正在尝试以下代码,但出现错误“找不到表或视图:df_joined_tables”。我该如何解决?提前谢谢!

%python
df_joined_tables = spark.table("df_joined_tables")

【问题讨论】:

    标签: python scala apache-spark pyspark apache-spark-sql


    【解决方案1】:

    要访问 Global Temp 视图,您需要在表名之前使用 global_temp 作为关键字。 在你的情况下试试这个:

    %python
    df_joined_tables = spark.table("global_temp.df_joined_tables")
    

    【讨论】:

      猜你喜欢
      • 2018-01-29
      • 1970-01-01
      • 1970-01-01
      • 2015-07-15
      • 1970-01-01
      • 2016-02-25
      • 1970-01-01
      • 2011-08-17
      • 1970-01-01
      相关资源
      最近更新 更多