在写入雪花之前保存 Spark 数据框

Question

我在 PySpark 中工作，在获得要写入 Snowflake 的最终输出表之前，我进行了一系列转换并应用了用户定义的函数。写入 Snowflake 的最终命令需要大约 25 分钟才能运行，因为它还在执行所有计算，因为 Spark 会延迟评估并且直到最后一次调用才会评估。 我想在之前的步骤中评估最终表，这样我就可以计算所有转换需要多长时间，然后分别计算写入 Snowflake 步骤需要多长时间。我如何将两者分开？我已经尝试过：

temp = final_df.show() 

temp.write.format(SNOWFLAKE_SOURCE_NAME).options(**sfOptions2) \
.option("dbtable","TEST_SPARK").save()

但我得到错误：

'NoneType' object has no attribute 'write'

然后用 collect()

temp = final_df.collect() 

temp.write.format(SNOWFLAKE_SOURCE_NAME).options(**sfOptions2) \
.option("dbtable","TEST_SPARK").save()

但我得到错误：

'list' object has no attribute 'write'

Answer 1

您的 temp 数据框的结果为 .show()，导致临时变量没有类型，只有 dataframe 对源有 .write 方法。

Try with below code:

temp = final_df
#view records from temp dataframe
temp.show()

temp.write.format(SNOWFLAKE_SOURCE_NAME).options(**sfOptions2) \
.option("dbtable","TEST_SPARK").save()

---

#collect collects the data as list and stores into temp variable
temp = final_df.collect() 

#list attributes doesn't have .write method
final_df.write.format(SNOWFLAKE_SOURCE_NAME).options(**sfOptions2) \
.option("dbtable","TEST_SPARK").save()

Update:

import time
start_time = time.time()
#code until show()
temp = final_df
#view records from temp dataframe
temp.show()
end_time = time.time()
print("Total execution time for action: {} seconds".format(end_time - start_time))

start_time_sfw = time.time()
#code until show()
final_df.write.format(SNOWFLAKE_SOURCE_NAME).options(**sfOptions2) \
.option("dbtable","TEST_SPARK").save()
end_time_sfw = time.time()
print("Total execution time for writing to snowflake: {} seconds".format(end_time_sfw - start_time_sfw))