如何使用 SPARK 将多个 parquet 文件转换为 TFrecord 文件？答案

【问题标题】：How to convert multiple parquet files into TFrecord files using SPARK?如何使用 SPARK 将多个 parquet 文件转换为 TFrecord 文件？
【发布时间】：2019-06-16 03:50:56
【问题描述】：

我想根据特定条件从大型 DataFrame 生成分层 TFrecord 文件，为此我使用write.partitionBy()。我也在 SPARK 中使用 tensorflow-connector，但这显然不能与 write.partitionBy() 操作一起使用。因此，除了尝试分两步工作之外，我还没有找到其他方法：

根据我的情况使用 partitionBy() 重新分区数据帧并将生成的分区写入 parquet 文件。
使用 tensorflow-connector 插件读取这些 parquet 文件以将其转换为 TFrecord 文件。

这是我无法有效完成的第二步。我的想法是读取 executors 上的各个 parquet 文件并立即将它们写入 TFrecord 文件。但这需要访问只能在驱动程序 (discussed here) 中完成的 SQLContext，因此不能并行。我想做这样的事情：

# List all parquet files to be converted
import glob, os
files = glob.glob('/path/*.parquet'))

sc = SparkSession.builder.getOrCreate()
sc.parallelize(files, 2).foreach(lambda parquetFile: convert_parquet_to_tfrecord(parquetFile))

我可以构造函数convert_parquet_to_tfrecord 来在执行程序上执行此操作吗？

我也试过在读取所有 parquet 文件时只使用通配符：

SQLContext(sc).read.parquet('/path/*.parquet')

这确实会读取所有 parquet 文件，但不幸的是不会读取到单个分区中。看起来原始结构丢失了，所以如果我想要将单个镶木地板文件的确切内容转换为 TFrecord 文件，这对我没有帮助。

还有其他建议吗？

【问题讨论】：

标签： apache-spark pyspark pyspark-sql parquet tfrecord

【解决方案1】：

如果我正确理解了您的问题，您希望将分区本地写入工作人员的磁盘上。

如果是这种情况，那么我建议您查看 spark-tensorflow-connector's 说明，了解如何操作。

这是您正在寻找的代码（如上面链接的文档中所述）：

myDataFrame.write.format("tfrecords").option("writeLocality", "local").save("/path")

另一方面，如果您担心效率，为什么要使用 pyspark？改用scala会更好。

【讨论】：

【解决方案2】：

试试 spark-tfrecord。

Spark-TFRecord 是一个类似于 spark-tensorflow-connector 的工具，但它执行 partitionBy。以下示例展示了如何对数据集进行分区。

import org.apache.spark.sql.SaveMode

// create a dataframe

val df = Seq((8, "bat"),(8, "abc"), (1, "xyz"), (2, "aaa")).toDF("number", "word")
val tf_output_dir = "/tmp/tfrecord-test"

// dump the tfrecords to files.
df.repartition(3, col("number")).write.mode(SaveMode.Overwrite).partitionBy("number").format("tfrecord").option("recordType", "Example").save(tf_output_dir)

更多信息可以在 Github 仓库： https://github.com/linkedin/spark-tfrecord

【讨论】：