问题标签 [parquet-mr]

我有一个使用 OOZIE 工作流 xml 并以序列文件格式（org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat）写入输出的 Map Reduce 作业，是否也有类似的东西可以保存为 Parquet 格式？，我不能在https://hadoop.apache.org/docs/stable/api/org/apache/hadoop/mapreduce/lib/output/package-summary.html下找到任何

还是我应该使用不同的方法？

请指教 。

谢谢

映射器：

减速器：

司机：

我必须在 parquet 文件上处理两个 map-reduce 程序
1）根据某些条件从 parquet 文件中识别少量记录并将它们的行号或某些唯一标识保存到一个地方
2）第二个 mapreduce 将需要读取这些唯一编号并从镶木地板文件中跳过这些记录

有没有什么机制，我可以从 parquet 创建一个唯一可区分的行 num

我需要在 Java 独立应用程序中读取镶木地板文件并发布到 Kafka。我有下面的代码来读取由 spark scala 应用程序生成的镶木地板文件。

java 应用程序是一个 Spring Boot 非 Web 应用程序。此代码在本地或 intelliJ IDE 中运行时工作正常，但是当相同的文件和代码被 dockerized 时，我收到以下错误。

我的 Dockerfile 供参考。

请帮助解决这个问题。

当使用 Spark 3.x 将 parquet 文件写入本地文件系统或 s3 或 HDFS 时，这些配置会影响 spark write parquet 的方式？

• spark.hadoop.parquet.dictionary.page.size
• spark.hadoop.parquet.block.size
• spark.hadoop.parquet.page.size
• spark.hadoop.parquet.enable.dictionary

问题目的

对 parquet 文件进行排序有很多好处：

• 使用文件元数据进行更有效的过滤
• 更有效的压缩率

这可能还有其他好处。互联网上有很多关于这个的讨论。由于这个原因，这个问题的讨论不是关于排序的原因。相反，这个问题的目的是谈论如何排序，在所有互联网链接中，解释最少（约30％）提到，而数据排序的挑战则根本没有提及。这个问题的目的是得到所有在这个领域的专家和经验丰富的朋友的帮助，并确定排序的最佳方法（基于成本和收益）。

Apache parquet 库简述

在开始讨论 Spark 之前，我将解释一下用于生成 parquet 文件的工具。当我们创建 parquet 文件时，该parquet-mr库（例如，我使用 Java，但它可能可以扩展到其他语言）同时写入磁盘和内存。这个库还有一个特性叫做getDataSize()返回文件在磁盘上完全关闭后的确切最终大小，所以我们可以在编写 parquet 文件时使用它来实现以下两个条件：

• 不要制作小尺寸的镶木地板文件（这对查询引擎不利）
• 所有 parquet 文件都可以生成具有一定的最小大小或固定大小（例如，每个文件 1 GB）

由于该库同时写入磁盘和内存，因此不允许对数据进行排序，除非所有数据都先在内存中排序，然后再交给库。（但这对于大量数据是不可能的。）我们还隐含地假设数据正在生成为我们打算存储的流。（在数据固定的情况下，这个问题说的问题就没有意义了，因为可以说整个数据一劳永逸，问题就结束了。但是我们假设有数据流，在这种情况下，重要的是有一个最佳的数据排序方式）

上面提到的 Apache parquet 库的一个优点是我们可以固定输出 parquet 文件的确切大小。在我看来，这是一个优势。因为，例如，如果我知道 Hadoop 块的大小等于 128 MB，parquet 行组的大小是 128 MB，我可以将 parquet 文件大小固定为 1 GB。然后我知道所有 parquet 文件将有 8 个块，HDFS 存储将得到最佳使用，所有 parquet 文件将是相同的。（因为在 HDFS 中，当块大小为 128 MB 时，较小的文件将占用相同数量的空间）这可能对每个人都没有好处，如果需要，我们很乐意有经验的人批评它。

Parquet 文件排序挑战

在我们开始之前的一点是，我们正在寻找永久的数据排序，因为我们将在接下来的数千次查询中使用它。几乎到目前为止，上述描述已经确定了排序的一些挑战，但我将在下面描述所有挑战：

1. Parquet 工具不允许您编写已排序的数据。所以一种方法是将所有数据保存在内存中，并在排序后将其交给 parquet 库以写入 parquet 文件。这种方法有两个缺点：1）不可能将所有数据都保存在内存中。2）由于所有数据都在内存中，parquet文件的大小是未知的，写入后可能小于或大于1GB或任意数量，失去了固定parquet大小的优势。
2. 假设我们想在并行过程中进行这种排序，而不是实时和流式地进行。这样一来，如果我们要使用parquet库，还是会有一个问题，就是必须把整个数据都拿到内存中进行排序，这是不可能的。因此，假设我们使用 Spark 之类的工具进行排序。我们在本节中给出的一个具体成本是集群资源用于排序，实际上每个数据被写入两次。（一次是parquet写入时间和一次排序）接下来的一点是，即使我们跳过这两种情况，在对数据进行排序之后，根据parquet文件中的其他列，该特定列的parquet压缩量和整个数据可能会发生变化和增加或减少。为此，parquet文件写入后，可能会创建小文件或更改固定大小（例如 1 GB）。不幸的是，Spark 并没有提供控制文件大小的方法（实际上可能无法实现），因此如果我们想恢复固定的文件大小，可能需要使用上面提到的链接等方法，这不会是免费的（导致除了消耗的集群资源之外多次写入文件并且确切的文件大小不会固定）：如何控制输出文件的大小

也许没有其他方法，唯一的方法就是上面提到的方法。在这种情况下，我很乐意让专家表达这个说明，以便其他人知道目前没有其他方法。

挑战总结

出于这个原因，我们通常在这些解决方案中观察到两种类型的问题：

1. 如何以合理的成本和时间进行排序（在流中）
2. 如何保持镶木地板文件的大小固定

出于这个原因，虽然到处都说排序非常好（而且我在网上和我自己的调查结果都表明它确实很有用），但完全没有提到它的方法和挑战。我请该领域有经验的专家朋友在这个方向上帮助我（希望它也能帮助其他人），如果在这个解释中遗漏了方法或要点，请说明。

对不起，如果由于我的英语语言薄弱，某些部分有错别字。谢谢。

我有一个大规模的分布式系统，它每天下载大量的大型 .csv 文件并对数据进行索引。可以说，我们的文件（file.csv）是： col1 col2 col3 user11 val12 val13 user21 val22 val23

然后我们逐行读取该文件并存储该文件中 user11 或 user12 的行所在位置的字节偏移量。例如：索引表 - user11 -> 1120-2130（字节偏移） user12 -> 2130-3545（字节偏移）

当有人说删除user11的数据时，我们参考这张表，下载并打开文件，删除文件中的字节偏移量。请注意，这个字节偏移量是整行的。

如何设计系统来处理镶木地板文件？Parquet 文件按列操作。要获得一整行的 10 列，我需要打 10 次电话吗？然后，形成一整行，计算字节，然后将它们存储在表中？然后，在删除时，我将不得不再次形成行然后删除字节？

其他选项是存储每列的字节偏移量并按列处理，但这会炸毁索引表。

如何以行方式有效地处理镶木地板文件？当前系统是 C# 中的后台作业。

我有一个镶木地板文件，其中有一列“FIXED_LEN_BYTE_ARRAY / UUID”，当我将它提供给 parquet-mr 库时，我得到了这个异常：

异常 - 由以下原因引起：org.apache.parquet.io.ParquetDecodingException：请求的架构与文件架构不兼容。不兼容的类型：必需的二进制文件

身份 (STRING) != 必需的 fixed_len_byte_array(16) 身份 (UUID)

在 org.apache.parquet.io.ColumnIOFactory$ColumnIOCreatorVisitor.incompatibleSchema(ColumnIOFactory.java:101)
在 org.apache.parquet.io.ColumnIOFactory$ColumnIOCreatorVisitor.visit(ColumnIOFactory.java:93)
在 org.apache.parquet.schema .PrimitiveType.accept(PrimitiveType.java:602)
at org.apache.parquet.io.ColumnIOFactory$ColumnIOCreatorVisitor.visitChildren(ColumnIOFactory.java:83)
at org.apache.parquet.io.ColumnIOFactory$ColumnIOCreatorVisitor.visit(ColumnIOFactory.java :57) 在 org.apache.parquet.io.ColumnIOFactory.getColumnIO(ColumnIOFactory.java:162) 在 org.apache.parquet.hadoop 的 org.apache.parquet.schema.MessageType.accept(MessageType.java:55)。 InternalParquetRecordReader.checkRead(InternalParquetRecordReader.java:135) ***

顺便提一句，

1. 我正在使用最新的 parquet-mr 库，即 1.12.0
2. 当我将相同的文件提供给 parquet cpp 库时，它能够对其进行解码。所以，我只想知道，parquet-mr 库 wrt UUID 中是否存在任何已知问题？

-开发