精通Spark数据科学
上QQ阅读APP看书，第一时间看更新
上一章目录下一章

2.2　内容登记

本章曾经提到过，数据采集是一个经常被忽略的领域，其实它的重要性不容低估。目前，我们已有了一条管道，可以从数据源采集数据，对采集过程进行调度，并将数据引导到选用的存储库中。但“故事”并不会就此结束。现在我们有了数据，接下来就要履行数据管理职责了。这就要引入内容登记。

对于采集到的数据，我们将建立一个与其相关的元数据索引。数据本身仍将定向存储（在示例中是HDFS），除此之外，我们将存储有关数据的元数据，这样就可以追踪接收到的数据，了解其基本信息。例如，何时收到它，它来自哪里，它有多大，它是什么类型的，等等。

2.2.1　选择和更多选择

选用哪种技术来存储元数据，主要是依赖知识和经验进行选择。对于元数据索引，至少需要具备以下特性。

  • 易于检索。
  • 可扩展。
  • 具有并行写入能力。
  • 支持冗余。

满足以上要求的技术方案有很多，例如，我们可以将元数据写入Parquet、存储在HDFS里、使用Spark SQL进行检索。不过，这里我们选用的是Elasticsearch，它能更好地满足以上条件。最值得注意的是，它可以通过REST API对元数据进行低延迟查询，这对于创建仪表盘非常有用。实际上，Elasticsearch还有一个优势是直接集成了Kibana，这意味着它可以为内容登记快速生成丰富的可视化。基于这些理由，我们将使用Elasticsearch。

2.2.2　随流而行

采用当前的NiFi管道流，让我们从“从URL获取GKG文件”（Fetch GKG files from URL）的输出里分出一点，以添加一组额外的步骤，允许我们捕获并存储元数据到Elasticsearch中。步骤如下。

  • 用元数据模型替换流的内容。
  • 捕获元数据。
  • 直接存储到Elasticsearch中。

上述步骤在NiFi里的操作结果界面如图2-11所示。

2-11

图2-11　将元数据保存到Elasticsearch

2.2.3　元数据模型

因此，第一步就是要定义元数据模型。要考虑的方面很多，但是，让我们先选择一个集合，以帮助解决前面讨论的几个关键点。这会为将来进一步增加数据提供良好的基础。我们先从简单的入手，使用以下3个属性。

  • 文件大小。
  • 采集日期。
  • 文件名。

以上属性就提供了接收到的文件的基本登记信息。

接下来，我们需要在NiFi流内部用这个新的元数据模型替换实际数据内容。有一个简单的方法：从模型中创建一个JSON模板文件。我们将它保存到本地磁盘上，并在FetchFile处理器中使用它，用骨架对象来替换流的内容。模板内容大致如下：

    {
      "FileSize": SIZE,
      "FileName": "FILENAME",
      "IngestedDate": "DATE"
    }

请注意，占位符名称（SIZE，FILENAME，DATE）代替了属性的值。这些会被逐个替换，替换顺序由ReplaceText处理器的序列控制，NiFi的表达语言通过正则表达式将占位符名称替换为适当的流属性，例如将Date替换为${now()}。

最后一个步骤是将新的元数据载荷输出到Elasticsearch中，NiFi通过一个叫作PutElasticsearch的处理器来进行这个操作。

下面是Elasticsearch中的一个元数据实体示例：

{
     "_index": "gkg",
     "_type": "files",
     "_id": "AVZHCvGIV6x-JwdgvCzW",
     "_score": 1,
     "source": {
        "FileSize": 11279827,
        "FileName": "20150218233000.gkg.csv.zip",
        "IngestedDate": "2016-08-01T17:43:00+01:00"
     }

现在我们已经拥有了收集和查询元数据的能力，并可以获取更多可用于分析的统计数据，包括以下内容。

  • 基于时间的分析，例如随着时间推移文件的大小变化。
  • 数据丢失，例如在时间轴上是否有数据空洞。

如果需要特定的分析，则NiFi元数据组件可以进行调整，以便提供相关数据点。事实上，可以构建一个分析平台来查看历史数据，如果当前数据中不存在元数据，则据此更新相应的索引。

2.2.4　Kibana仪表盘

前文已经多次提到Kibana。现在Elasticsearch中有一个元数据的索引，我们可以使用这个工具来对一些分析进行可视化。这个简要介绍的目的是证明我们可以立即对数据进行建模和可视化。要了解在更复杂的场景中如何使用Kibana，请阅读第9章的相关内容。在这个简单的示例中，我们要完成以下步骤。

  • 在“设置”选项卡中为GDELT元数据添加Elasticsearch索引。
  • 在“发现”选项卡下选定“文件大小”。
  • 为“文件大小”选择“可视化”。
  • 将“聚合字段”更改为“范围”。
  • 输入“范围”的值。

生成的图表展示了文件大小的分布情况，如图2-12所示。

2-12

图2-12　文件大小分布情况

至此，我们可以自由地创建新的可视化，甚至可以构建一个功能齐全的仪表盘，用来监控文件采集的状态。通过增加从NiFi写入Elasticsearch的元数据的多样性，我们可以在Kibana中探索更多的领域，甚至获得一些基于采集的可行的见解，从而开始我们的数据科学旅程。

现在，我们已经有了一个功能齐全的数据管道，它能提供实时的数据流，那该如何确保正在接收的载荷数据的质量？下面我们来看看有什么方法。

上一章目录下一章