使用 Apache NiFi 实时传输源数据处理

使用 Apache NiFi 实时流式传输的 GTFS

为了便于引入 GTFS 实时数据，我添加了一个处理器，可将格式化的 ProtoBuf 数据转换为 JSON 的 GTFS（一般传输馈送规范）。这是使用标准的谷歌API来实现这一点。您可以在此处查看协议缓冲区架构。

我们将能够获得行程更新的数据，包括延误和服务警报，包括改变的路线和车辆位置，可以有位置和拥堵信息。这是相同的信息，这些公共交通系统提供谷歌地图。

用于访问 GTFS 数据的示例 NiFi 流

首先，我们添加我的新NiFi处理器，你可以得到作为一个源，并建立NAR。或下载我的新 NiFi NAR 存档的发布版本之一。这是阿尔法级代码完成得非常快。如果您发现错误，请报告并建议更新。

将其添加到画布后，可以更改名称和日程安排。

当前唯一的设置是 GTFS 资源的 URL（可能需要来自某些运输 API 的密钥）。

运行后，它将快速返回包含 JSON 格式的 GTFS 结果版本以及包含 GTFS 信息的某些属性的流文件。

JSON 代码有很多数组和子节，不是很平坦。我们可以将原始文件转储到 Hive，并将其作为 JSON 文档进行查询。或者，我们可以解析它与拼音JSON，评估JSONPath，斯普利特Json，或各种其他选项。

如前所述，主要部分是tripUpdate，车辆和警报。 我将对此进行分析，并将其用于实时 Slack、Kafka、HBase 和 Kudu 消息传递。对于有卡车或人员尝试导航系统的公司，此数据可能变得至关重要。

我的第一个用例是查看我是否可以在紧急情况或事件期间实时发现这些数据以帮助社区。非营利机构将能够有关于可用运输的实时状态报告。您必须引入、清理和转换数据，从而成为知识，从而增强社区的能力。

和我一起到波多黎各参加NetHope全球峰会。

源代码

https://github.com/tspannhw/gtfs/

要在 Apache NiFi 1.9.2 中安装的纳尔版本

https://github.com/tspannhw/gtfs/releases

资源

• https://cdn.mbta.com/realtime/TripUpdates.pb
• https://cdn.mbta.com/realtime/VehiclePositions.pb
• https://cdn.mbta.com/realtime/Alerts.pb
• https://www.mbta.com/developers/gtfs-realtime
• https://github.com/CUTR-at-USF/gtfs-realtime-validator