从集群将整数/字符串写入 pyspark 中的文本文件

Posted 2023-04-15

【中文标题】从集群将整数/字符串写入 pyspark 中的文本文件

【英文标题】：Writing integer/string to a text file in pyspark from a cluster

【发布时间】：2020-03-23 01:26:42

【问题描述】：

我正在使用 EMR 阶跃函数来分析数据。 我想存储分析数据帧的计数，以决定是否可以将其保存为 csv 或 parquet。我更喜欢 CSV，但如果尺寸太大，我将无法下载它并在我的笔记本电脑上使用它。 我使用count() 方法将它存储到一个int 变量limit 当我尝试使用以下代码时：

coalesce(1).write.format("text").option("header", "false").mode("overwrite").save("output.txt")

上面写着：

int 没有任何名为 write 的属性

有没有办法将整数或字符串写入文件，以便我可以在我的 s3 存储桶中打开它并在 EMR 步骤运行后进行检查？

更新： 我尝试了@Shu 建议的数据框方法，但出现以下错误。

原因：org.apache.spark.SparkException：作业因阶段而中止 失败：阶段 13.0 中的任务 0 失败 4 次，最近一次失败： 在 13.0 阶段丢失任务 0.3（TID 19396，ip-10-210-13-34.ec2.internal， 执行程序 11）：org.apache.spark.SparkException：任务失败，而 写行。在 org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:257) 在 org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:170)

这可能是什么根本原因？

【参考方案1】：

您可以parallelize int 变量创建一个rdd，然后使用.saveAsTextFile 写入HDFS

df.show()
#+---+
#| _1|
#+---+
#|  a|
#|  b|
#+---+
limit=df.count()
spark.sparkContext.parallelize([limit]).coalesce(1).saveAsTextFile("<path>")

#content of file
#cat <path>/part-00000
#2 

---

其他方法是从count variable 创建dataframe，然后以csv 格式写入标题false。

from pyspark.sql.types import *
spark.createDataFrame(spark.sparkContext.parallelize([limit]),IntegerType()).coalesce(1).write.format("csv").option("header", "false").mode("overwrite").save("<path>")

#or in text format
spark.createDataFrame(spark.sparkContext.parallelize([limit]),StringType()).coalesce(1).write.format("text").mode("overwrite").save("<path>")

#cat part-*
#2

【讨论】：

我试图将其转换为数据帧并写入，但它一直失败。我没有使用IntegerType() 方法。非常感谢舒！