写入 HDFS - hive 时，如何控制 Spark 流中的行数和/或输出文件大小？答案

【问题标题】：How can I control number of rows and/or output file size in Spark streaming when writing to HDFS - hive?写入 HDFS - hive 时，如何控制 Spark 流中的行数和/或输出文件大小？
【发布时间】：2018-10-18 00:53:35
【问题描述】：

使用 spark 流来读取和处理来自 Kafka 的消息并写入 HDFS - Hive。由于我希望避免创建许多会向文件系统发送垃圾邮件的小文件，因此我想知道是否有办法确保最小的文件大小和/或强制文件中的输出行数最少的能力，除了超时。谢谢。

【问题讨论】：

您是否考虑过使用Kafka Connect HDFS 连接器从Kafka 主题写入HDFS？与 Kafka Streams 或 KSQL 一起执行任何所需的处理。
Kafka KSQL/Streams 可以对 JSON 文件进行转换吗？在将 JSON 写入 HDFS-hive 之前，我需要能够处理 JSON 并执行一些转换。谢谢。
是的，他们可以。下面是一个简单的例子：confluent.io/blog/…
Kafka 消息只是字节，而不是文件。您可以编写任何消费者来解析 JSON 字符串、操作、写入单独的“丰富”或“过滤”主题，然后将其接收到 HDFS

标签： scala apache-spark hadoop apache-kafka spark-streaming

【解决方案1】：

据我所知，没有办法控制输出文件中的行数。但是您可以控制输出文件的数量。

控制它并考虑您的数据集大小可能会帮助您满足您的需求，因为您可以计算输出中每个文件的大小。您可以使用coalesce 和repartition 命令来做到这一点：

df.coalesce(2).write(...)
df.repartition(2).write(...)

它们都用于创建作为参数给出的分区数。所以如果你设置 2，你的输出应该有 2 个文件。

不同之处在于repartition 可以增加和减少分区，而coalesce 只能减少。

另外，请记住，repartition 执行完全洗牌以在分区之间平均分配数据，这可能会耗费资源和时间。另一方面，coalesce 不执行完全 shuffle，而是结合现有分区。

你可以在这个其他答案here中找到一个很棒的解释

【讨论】：

谢谢。假设我有一个有 100 个分区的 Kafka 主题，在你的 RDD 中只有（比如说）10 个分区所以你只输出 10 个文件是可以接受的做法吗？此外，由于我们正在讨论流式传输，因此文件最终需要关闭，什么时候会发生这种情况？对于每个 RDD？
关于实践，这取决于您的数据集大小。每个分区必须具有适当的大小。这应该是 1kb 也不是 1500Tb。