SparkSQL中臨時表和資料庫表的可見性，是否可以從通常的jdbc查詢對臨時表進行嵌套查詢-有解無憂

我有一個DataFrame臨時表

val dailySummariesDfVisualize =
      dailySummariesDf
    .orderBy("event_time").registerTempTable("raw")

我可以使用 Spark SQL 從中提取一些資訊：

val df = sqlContext.sql("SELECT * FROM raw")
df.show()

并且輸出有效。然后我想對 JDBC 資料庫查詢中的臨時表進行嵌套查詢，如下所示：

val dailySensorData =
getDFFromJdbcSource(SparkSession.builder().appName("test").master("local").getOrCreate(), 
          s"SELECT *  FROM values WHERE time in (SELECT event_time FROM raw) limit 1000000")
           .persist(StorageLevel.MEMORY_ONLY_SER)
dailySensorData.show(400, false)

在這里我得到了例外：

org.postgresql.util.PSQLException: ERROR: relation "raw" does not exist

如果我嘗試在內部執行sqlContext.sql()這樣的

val df = sqlContext.sql("SELECT * FROM values WHERE time in (SELECT event_time FROM raw)")
df.show()

我得到：

org.apache.spark.sql.AnalysisException: Table or view not found: values; line 1 pos 14;
'Project [*]
 - 'Filter 'time IN (list#4967 [])
   :   - 'Project ['event_time]
   :      - 'UnresolvedRelation [raw]
    - 'UnresolvedRelation [values]

  at org.apache.spark.sql.catalyst.analysis.package$AnalysisErrorAt.failAnalysis(package.scala:42)
  at org.apache.spark.sql.catalyst.analysis.CheckAnalysis.$anonfun$checkAnalysis$1(CheckAnalysis.scala:106)

就像值（真正的 jdbc 表）和原始（臨時表）一樣，它都不可見。如何在嵌套查詢中使用臨時表？

UPD

根據 mazaneicha 我試過（在這里檢索所有值，因為無法使用嵌套查詢限制它們）：

val dailySummariesDfVisualize =
      dailySummariesDf
    .orderBy("event_time").createOrReplaceTempView("raw") 

val dailySensorData =
      getDFFromJdbcSource(SparkSession.builder().appName("test").master("local").getOrCreate(), 
      s"SELECT *  FROM values").createOrReplaceTempView("values")     

val df = sqlContext.sql("SELECT * FROM values WHERE time in (SELECT event_time FROM raw)")
df.explain(true)

這是邏輯計劃：

= Parsed Logical Plan ==
'Project [*]
 - 'Filter 'time IN (list#5475 [])
   :   - 'Project ['event_time]
   :      - 'UnresolvedRelation [raw]
    - 'UnresolvedRelation [values]

== Analyzed Logical Plan ==
devicename: string, value: double, time: timestamp, coffee_machine_id: string, digital_twin_id: string, write_time: timestamp
Project [devicename#5457, value#5458, time#5459, coffee_machine_id#5460, digital_twin_id#5461, write_time#5462]
 - Filter time#5459 IN (list#5475 [])
   :   - Project [event_time#4836]
   :      - SubqueryAlias raw
   :         - Sort [event_time#4836 ASC NULLS FIRST], true
   :            - Relation[event_type#4835,event_time#4836,event_payload#4837,coffee_machine_id#4838,digital_twin_id#4839] JDBCRelation((SELECT *  FROM events WHERE (event_time > '2021-03-31'  or event_time < '2021-03-30') and event_type != 'Coffee_Capsule_RFID_Event' and event_type!='Coffee_Cup_RFID_Event' limit 2000000) SPARK_GEN_SUBQ_48) [numPartitions=1]
    - SubqueryAlias values
       - Relation[devicename#5457,value#5458,time#5459,coffee_machine_id#5460,digital_twin_id#5461,write_time#5462] JDBCRelation((SELECT *  FROM values) SPARK_GEN_SUBQ_65) [numPartitions=1]

== Optimized Logical Plan ==
Join LeftSemi, (time#5459 = event_time#4836)
:- Relation[devicename#5457,value#5458,time#5459,coffee_machine_id#5460,digital_twin_id#5461,write_time#5462] JDBCRelation((SELECT *  FROM values) SPARK_GEN_SUBQ_65) [numPartitions=1]
 - Project [event_time#4836]
    - Relation[event_type#4835,event_time#4836,event_payload#4837,coffee_machine_id#4838,digital_twin_id#4839] JDBCRelation((SELECT *  FROM events WHERE (event_time > '2021-03-31'  or event_time < '2021-03-30') and event_type != 'Coffee_Capsule_RFID_Event' and event_type!='Coffee_Cup_RFID_Event' limit 2000000) SPARK_GEN_SUBQ_48) [numPartitions=1]

== Physical Plan ==
SortMergeJoin [time#5459], [event_time#4836], LeftSemi
:- *(2) Sort [time#5459 ASC NULLS FIRST], false, 0
:   - Exchange hashpartitioning(time#5459, 200), true, [id=#1219]
:      - *(1) Scan JDBCRelation((SELECT *  FROM values) SPARK_GEN_SUBQ_65) [numPartitions=1] [devicename#5457,value#5458,time#5459,coffee_machine_id#5460,digital_twin_id#5461,write_time#5462] PushedFilters: [], ReadSchema: struct<devicename:string,value:double,time:timestamp,coffee_machine_id:string,digital_twin_id:str...
 - *(4) Sort [event_time#4836 ASC NULLS FIRST], false, 0
    - Exchange hashpartitioning(event_time#4836, 200), true, [id=#1224]
       - *(3) Scan JDBCRelation((SELECT *  FROM events WHERE (event_time > '2021-03-31'  or event_time < '2021-03-30') and event_type != 'Coffee_Capsule_RFID_Event' and event_type!='Coffee_Cup_RFID_Event' limit 2000000) SPARK_GEN_SUBQ_48) [numPartitions=1] [event_time#4836] PushedFilters: [], ReadSchema: struct<event_time:timestamp>

uj5u.com熱心網友回復：

根據 mazaneicha 的建議，我能夠通過從 DataFramw Rows 生成 scala 中的 where 子句來解決這個問題，與我從中進行提取查詢的資料相比，這些子句并不多：

var collectedString = scala.collection.mutable.MutableList[String]()

for (row <- dailySummariesDfVisualize.collectAsList())
  {
      println(row(1))
      val start = row(1)
      val end = row(5)
      val timeSelection = s" time > ' ${start}' and  time < '${end}'"
      collectedString =timeSelection    
  }

val whereClause = collectedString.mkString(" or ")
println(whereClause)

val dailySensorData =
      getDFFromJdbcSource(SparkSession.builder().appName("test").master("local").getOrCreate(), 
      s"SELECT *  FROM values WHERE " whereClause " limit 1000000")
       .persist(StorageLevel.MEMORY_ONLY_SER)    

dailySensorData.show(400, false)

它以可接受的性能輸出我實際需要的東西。

格式化的 whereClause 輸出類似于：

time > ' 2021-03-24 07:06:34.0' and  time < '2021-03-24 07:08:34.0' or  time > ' 2021-03-24 07:07:41.0' and  time < '2021-03-24 07:09:41.0' or  time > ' 2021-03-24 07:07:43.0' and  time < '2021-03-24 07:09:43.0'

等等

轉載請註明出處，本文鏈接：https://www.uj5u.com/qiye/387350.html

標籤：斯卡拉阿帕奇火花 apache-spark-sql 阿帕奇齐柏林飞艇

上一篇：如何從使用JDBC檢索的資料庫表中獲取DataFrame，該表由多個日期范圍切割，另一個DataFrame中的每行一個日期范圍？

下一篇：Gatling：獲取JSON陣列中第一個物件的第一個元素