我在 Jupyter 筆記本中有一個 Spark SQL DataFrame 作為output_df1. 我想定義一個函式如下：
def output_agg(output_table_1):
output_agg_1 = spark.sql(f"""
select * from {output_table_1}
""")
return output_agg_1
當我打電話時output_agg(output_df1)，我收到以下錯誤：
Py4JJavaError Traceback (most recent call last)
/opt/spark/python/lib/pyspark.zip/pyspark/sql/utils.py in deco(*a, **kw)
62 try:
---> 63 return f(*a, **kw)
64 except py4j.protocol.Py4JJavaError as e:
/opt/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
327 "An error occurred while calling {0}{1}{2}.\n".
--> 328 format(target_id, ".", name), value)
329 else:
Py4JJavaError: An error occurred while calling o110.sql.
: org.apache.spark.sql.catalyst.parser.ParseException:
mismatched input '[' expecting <EOF>
你能幫忙提供正確的語法嗎？
uj5u.com熱心網友回復：
spark sqlselect需要給表/臨時表。先將dataframe注冊為臨時表，然后再執行SQL陳述句。
output_df1.createOrReplaceTempView('output_table')
def output_agg(output_table_1):
output_agg_1 = spark.sql(f"""
select * from {output_table_1}
""")
return output_agg_1
output_agg('output_table')
我無法對已完成的問題添加評論。學習 Spark SQL 的資訊是 Spark SQL Reference。https://spark.apache.org/docs/latest/sql-ref.html
uj5u.com熱心網友回復：
在傳遞給之前列印 SQL 查詢spark.sql并檢查 SQL 查詢是否看起來不錯。另外，分享有問題的 SQL 查詢。
def output_agg(output_table_1):
query = f"""select * from {output_table_1}"""
print(query)
output_agg_1 = spark.sql(query)
return output_agg_1
如果 SQL 查詢看起來不錯，那么可能的問題可能是表未在 spark 中注冊。
正如評論中所討論的，既然你想聯合多個 dfs，你可以做這樣的事情
from functools import reduce
from pyspark.sql import DataFrame
dfs_list = [output_df1, output_df2, output_df3, output_df4]
df_combined = reduce(DataFrame.unionAll, dfs_list)
注意：確保所有 df 中的列順序相同
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/378885.html