读Kafka的书，写拼花地板中的hdfs

Question

我是大数据生态系统的新手，有点起步。

我读过几篇关于使用spark流媒体阅读Kafka主题的文章，但我想知道是否可以使用spark作业而不是流媒体阅读Kafka主题？如果是的话，你们能帮我指出一些可以让我开始学习的文章或代码片段吗。

问题的第二部分是以拼花格式向hdfs写信。一旦我读了Kafka的书，我想我会有一个rdd。将此rdd转换为数据帧，然后将数据帧写入拼花文件。这是正确的方法吗。

感谢您的帮助。

谢谢

Answer 1

使用Kafka流。SparkStreaming用词不当（它是引擎盖下的小批量，至少高达2.2）。

https://eng.verizondigitalmedia.com/2017/04/28/Kafka-to-Hdfs-ParquetSerializer/

Answer 2

关于这个话题，你已经有了几个很好的答案。

只是想释放压力-小心直接流到拼花地板桌上。当拼花地板行组大小足够大（为简单起见，您可以说文件大小应该在64-256Mb左右），以利用字典压缩、bloom过滤器等（一个拼花文件中可以有多个行块，并且通常每个文件中都有多个行块；虽然行块不能跨越多个拼花文件），拼花的性能就会大放异彩

如果您直接流式处理到一个拼花地板表，那么很可能会得到一堆很小的拼花地板文件（取决于Spark流式处理的小批量大小和数据量）。查询此类文件可能非常慢。例如，拼花可能需要读取所有文件的头来协调模式，这是一个很大的开销。如果是这种情况，您将需要一个单独的进程，例如，作为一种解决方法，该进程将读取旧文件，并将其写入“合并”（这不是一个简单的文件级合并，一个进程实际上需要读取所有拼花数据并溢出更大的拼花文件）。

这种变通方法可能会扼杀数据“流”的最初目的。你也可以看看其他技术，比如Apache Kudu、Apache Kafka、Apache Druid、Kinesis等，它们可以在这里更好地工作。

更新：自从我发布了这个答案，现在这里有了一个新的强大玩家——三角洲湖。https://delta.io/如果你习惯拼花，你会发现三角洲非常有吸引力（实际上，三角洲建立在拼花层元数据之上）。三角洲湖提供：

Spark上的ACID事务：

• 可序列化的隔离级别确保读者永远不会看到不一致的数据
• 可扩展元数据处理：利用Spark的分布式处理能力，轻松处理具有数十亿文件的PB级表的所有元数据
• 流和批处理统一：Delta Lake中的一个表既是一个批处理表，也是一个流源和流汇。流式数据摄取、批量历史回填、交互式查询都是现成的
• 模式强制：自动处理模式变化，以防止在摄取过程中插入错误记录
• 时间旅行：数据版本控制支持回滚、完整的历史审核跟踪和可复制的机器学习实验
• 升级和删除：支持合并、更新和删除操作，以支持复杂的用例，如更改数据捕获、缓慢更改维度（SCD）操作、流式升级等

Answer 3

要以拼花格式从Kafka读取数据并将其写入HDFS，请使用Spark Batch job而不是流式处理，您可以使用Spark结构化流式处理。

结构化流是一种基于Spark SQL引擎构建的可扩展、容错的流处理引擎。您可以使用与对静态数据进行批处理计算相同的方式来表示流计算。Spark SQL引擎将负责以增量方式连续运行它，并在流式数据不断到达时更新最终结果。您可以使用Scala、Java、Python或R中的Dataset/DataFrame API来表示流聚合、事件时间窗口、流到批连接等。计算是在同一个优化的Spark SQL引擎上执行的。最后，该系统通过检查点和预写日志确保端到端的精确一次容错。简而言之，结构化流提供了快速、可扩展、容错、端到端的一次流处理，用户无需对流进行推理。

它作为Kafka的内置源提供，即我们可以从Kafka轮询数据。它与Kafka broker版本0.10.0或更高版本兼容。

为了在批处理模式下从Kafka提取数据，您可以为定义的偏移范围创建数据集/数据帧。

// Subscribe to 1 topic defaults to the earliest and latest offsets
val df = spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

// Subscribe to multiple topics, specifying explicit Kafka offsets
val df = spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1,topic2")
  .option("startingOffsets", """{"topic1":{"0":23,"1":-2},"topic2":{"0":-2}}""")
  .option("endingOffsets", """{"topic1":{"0":50,"1":-1},"topic2":{"0":-1}}""")
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

// Subscribe to a pattern, at the earliest and latest offsets
val df = spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribePattern", "topic.*")
  .option("startingOffsets", "earliest")
  .option("endingOffsets", "latest")
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

源中的每一行都有以下架构：

| Column           | Type          |
|:-----------------|--------------:|
| key              |        binary |
| value            |        binary |
| topic            |        string |
| partition        |           int |
| offset           |          long |
| timestamp        |          long |
| timestampType    |           int |

现在，要以拼花格式将数据写入HDFS，可以编写以下代码：

df.write.parquet("hdfs://data.parquet")

有关Spark结构化流媒体Kafka的更多信息，请参阅以下指南-Kafka集成指南

我希望这有帮助！