使用spark解析NiFi数据包

Question

我正在使用 Apache NiFi 和 Apache Spark 为大学做一个小项目。我想使用 NiFi 创建一个从 HDFS 读取 TSV 文件的工作流，并使用 Spark Streaming 我可以处理文件并将我需要的信息存储在 MySQL 中。我已经在 NiFi 中创建了我的工作流，并且存储部分已经在工作了。问题是我无法解析 NiFi 包，所以我可以使用它们。

文件包含这样的行：

linea1File1 TheReceptionist 653 Entertainment   424 13021   4.34    1305    744 DjdA-5oKYFQ NxTDlnOuybo c-8VuICzXtU

其中每个 space 是一个制表符 ("\t")

这是我在 Spark 中使用 Scala 的代码：

 val ssc = new StreamingContext(config, Seconds(10))
 val packet = ssc.receiverStream(new NiFiReceiver(conf, StorageLevel.MEMORY_ONLY))
 val file = packet.map(dataPacket => new String(dataPacket.getContent, StandardCharsets.UTF_8))

到这里为止，我可以在一个字符串中获取我的整个文件（7000 多行）...不幸的是，我无法将该字符串分成几行。我需要按行获取整个文件，这样我就可以在一个对象中解析它，对其应用一些操作并存储我想要的内容

谁能帮我解决这个问题？

Answer 1

每个数据包都将是来自NiFi的一个流文件的内容，所以如果NiFi从HDFS中获取一个有很多行的TSV文件，所有这些行将在一个数据包中。

很难说没有看到你的 NiFi 流程，但你可能会使用行数为 1 的 SplitText 在它开始流式传输之前在 NiFi 中拆分你的 TSV。