在2019年最后一个季度，我使用Spark开发了元数据驱动的提取引擎。框架/库具有多种模式，可以满足多种源和目标组合。例如，有两种模式可用于将平面文件加载到云存储（一种模式用于将数据加载到AWS S3，另一种模式用于将数据加载到Azure Blob）。
随着数据加载理念从“提取转换加载”（ETL）变为“ 提取加载转换（ETL）”，这样的框架非常有用，因为它减少了设置提取作业所需的时间。
任何摄取引擎的一个重要方面是要知道从给定的源读取了多少记录并将其写入目的地。通常，想到的方法是对DataFrame执行计数操作已加载。这将为我们提供从源加载的记录数。在将数据写入存储的情况下，我们需要将数据加载到另一个DataFrame中并对其进行计数。
但是，对DataFrame进行计数操作可能会很昂贵。有其他选择吗？事实证明，有。另一种方法是注册Spark事件。这是通过从SparkListener类扩展我们的类并覆盖OnStageCompleted方法或OnTaskEnd方法（取决于我们要执行的操作）来完成的。
每当活动完成时，Spark都会OnStageCompleted在已注册的侦听器上调用该方法。这种方法使我们能够跟踪执行时间和执行者占用的CPU时间。任务完成后，Spark会调用OnTaskEndSpark侦听器上的方法。此方法可用于确定读取和写入的记录数。
为了跟踪执行时间，读取的记录数和写入的记录数，我在本文中介绍了一些帮助程序类。要注册活动完成任务，您需要从StageCompletedEventConsumer特征中派生您的课程。要注册读取计数，您需要从RecordsLoadedEventConsumer特征导出类。
要注册写计数，您需要从RecordsWrittenEventConsumer特征中派生您的类。从给定的特征派生类之后，您需要将该类添加到各自的管理器类中。事件发生时，Spark将调用管理器类，该类又将通知所有已注册的侦听器。
读取事件呈现了一个有趣的情况。当Spark读取数据时，read事件将被调用两次：第一次是在读取第一条记录之后，第二次是在加载所有记录之后。换句话说，使用两个值调用读取事件侦听器。第一次，读取的记录的值是一个。第二次，读取的记录的值是数据集中的记录数。