Presto Event Listener开发

简介

同Hive Hook一样，Presto也支持自定义实现Event Listener，用于侦听Presto引擎执行查询时发生的事件，并作出相应的处理。我们可以利用该功能实现诸如自定义日志记录、调试和性能分析插件，帮助我们更好的运维Presto集群。但是不同于Hive Hook的是，在Presto集群中，一次只能有一个Event Listener处于活动状态。

Event Listener作为Plugin监听以下事件：

Query Creation(查询建立相关信息)
Query completion (success or failure)(查询执行相关信息，包含成功查询的细节信息，失败查询的错误码等信息)
Split completion (success or failure)(split执行信息，同理包含成功和失败的细节信息)

了解Hook及Listener模式的朋友对于其步骤应该很清楚了，我们只需要：

实现Presto Event Listener和EventListenerFactory接口。
正确的打包我们的jar。
部署，放到Presto指定目录，修改配置文件。

接口

实现EventListener，该类是我们的核心逻辑所在，供包含上面所说的三个事件：

public interface EventListener
{
	//query创建的详细信息
    default void queryCreated(QueryCreatedEvent queryCreatedEvent)
    {
    }
	//query执行的详细信息
    default void queryCompleted(QueryCompletedEvent queryCompletedEvent)
    {
    }
	//split执行的详细信息
    default void splitCompleted(SplitCompletedEvent splitCompletedEvent)
    {
    }
}

实现EventListenerFactory创建我们自己实现的EventListener
实现Plugin接口，实现getEventListenerFactories()方法，获取我们自己实现的EventListenerFactory
添加配置信息，为etc/event-listener.properties。其中event-listener.name为必备属性，其他属性为我们plugin所需要的信息。

示例

由于集群运维的需要，先需要将用户的查询历史、查询花费的时间等信息进行统计，以便于后续对各个业务的查询进行优先级分级和评分，方便后续Presto集群稳定性易用性的维护。这里给出一个简单的将这些信息存储到Mysql数据库的样例。

Maven Pom


      com.facebook.presto
      presto-spi
      0.220
      compile

QueryEventListenerFactory

public class QueryEventListenerFactory implements EventListenerFactory {

  @Override
  public String getName() {
    return "query-event-listener";
  }

  @Override
  public EventListener create(Map config) {
    if (!config.containsKey("jdbc.uri")) {
      throw new RuntimeException("/etc/event-listener.properties file missing jdbc.uri");
    }
    if (!config.containsKey("jdbc.user")) {
      throw new RuntimeException("/etc/event-listener.properties file missing jdbc.user");
    }
    if (!config.containsKey("jdbc.pwd")) {
      throw new RuntimeException("/etc/event-listener.properties file missing jdbc.pwd");
    }

    return new QueryEventListener(config);
  }
}

QueryEventPlugin

public class QueryEventPlugin implements Plugin {

  @Override
  public Iterable getEventListenerFactories() {
    EventListenerFactory listenerFactory = new QueryEventListenerFactory();
    return Arrays.asList(listenerFactory);
  }
}

QueryEventListener

public class QueryEventListener implements EventListener {

  private Map config;
  private Connection connection;

  public QueryEventListener(Map config) {
    this.config = new HashMap<>();
    this.config.putAll(config);
    init();
  }

  private void init() {
    try {
      if (connection == null || !connection.isValid(10)) {
 Class.forName("com.mysql.jdbc.Driver");
 connection = DriverManager
     .getConnection(config.get("jdbc.uri"), config.get("jdbc.user"), config.get("jdbc.pwd"));
      }
    } catch (SQLException | ClassNotFoundException e) {
      e.printStackTrace();
    }
  }

  @Override
  public void queryCreated(QueryCreatedEvent queryCreatedEvent) {
  }

  @Override
  public void queryCompleted(QueryCompletedEvent queryCompletedEvent) {
    String queryId = queryCompletedEvent.getmetadata().getQueryId();
    String querySql = queryCompletedEvent.getmetadata().getQuery();
    String queryState = queryCompletedEvent.getmetadata().getQueryState();
    String queryUser = queryCompletedEvent.getContext().getUser();
    long createTime = queryCompletedEvent.getCreateTime().toEpochMilli();
    long endTime = queryCompletedEvent.getEndTime().toEpochMilli();
    long startTime = queryCompletedEvent.getExecutionStartTime().toEpochMilli();
    //insert into query execution table

    long analysisTime = queryCompletedEvent.getStatistics().getAnalysisTime().orElse(Duration.ZERO)
 .toMillis();
    long cpuTime = queryCompletedEvent.getStatistics().getCpuTime().toMillis();
    long queuedTime = queryCompletedEvent.getStatistics().getQueuedTime().toMillis();
    long wallTime = queryCompletedEvent.getStatistics().getWallTime().toMillis();
    int completedSplits = queryCompletedEvent.getStatistics().getCompletedSplits();
    double cumulativeMemory = queryCompletedEvent.getStatistics().getCumulativeMemory();
    long outputBytes = queryCompletedEvent.getStatistics().getOutputBytes();
    long outputRows = queryCompletedEvent.getStatistics().getOutputRows();
    long totalBytes = queryCompletedEvent.getStatistics().getTotalBytes();
    long totalRows = queryCompletedEvent.getStatistics().getTotalRows();
    long writtenBytes = queryCompletedEvent.getStatistics().getWrittenBytes();
    long writtenRows = queryCompletedEvent.getStatistics().getWrittenRows();
	//insert into query info table
	
    queryCompletedEvent.getFailureInfo().ifPresent(queryFailureInfo -> {
      int code = queryFailureInfo.getErrorCode().getCode();
      String name = queryFailureInfo.getErrorCode().getName();
      String failureType = queryFailureInfo.getFailureType().orElse("").toUpperCase();
      String failureHost = queryFailureInfo.getFailureHost().orElse("").toUpperCase();
      String failureMessage = queryFailureInfo.getFailureMessage().orElse("").toUpperCase();
      String failureTask = queryFailureInfo.getFailureTask().orElse("").toUpperCase();
      String failuresJson = queryFailureInfo.getFailuresJson();
      // insert into failed query table
    });
  }


  @Override
  public void splitCompleted(SplitCompletedEvent splitCompletedEvent) {
    long createTime = splitCompletedEvent.getCreateTime().toEpochMilli();
    long endTime = splitCompletedEvent.getEndTime().orElse(Instant.MIN).toEpochMilli();
    String payload = splitCompletedEvent.getPayload();
    String queryId = splitCompletedEvent.getQueryId();
    String stageId = splitCompletedEvent.getStageId();
    long startTime = splitCompletedEvent.getStartTime().orElse(Instant.MIN).toEpochMilli();
    String taskId = splitCompletedEvent.getTaskId();
    long completedDataSizeBytes = splitCompletedEvent.getStatistics().getCompletedDataSizeBytes();
    long completedPositions = splitCompletedEvent.getStatistics().getCompletedPositions();
    long completedReadTime = splitCompletedEvent.getStatistics().getCompletedReadTime().toMillis();
    long cpuTime = splitCompletedEvent.getStatistics().getCpuTime().toMillis();
    long queuedTime = splitCompletedEvent.getStatistics().getQueuedTime().toMillis();
    long wallTime = splitCompletedEvent.getStatistics().getWallTime().toMillis();
    //insert into stage info table
  }

}

打包

Presto使用服务提供者接口（SPI）来扩展Presto。Presto使用SPI加载连接器，功能，类型和系统访问控制。SPI通过元数据文件加载。我们还需要创建src/main/resources/meta-INF/services/com.facebook.presto.spi.Plugin元数据文件。该文件应包含我们插件的类名如： com.ji3jin.presto.listener.QueryEventListener
执行mvn clean install打包

部署

创建配置文件etc/event-listener.properties

event-listener.name=query-event-listener

jdbc.uri=jdbc:mysql://localhost:3306/presto_monitor
jdbc.user=presto
jdbc.pwd=presto123

在presto根目录下创建query-event-listener目录，名称与我们上面event listener的name一致
将我们的jar包和mysql connector的jar包拷贝到上面创建的目录
重新启动Presto服务即可

好了，现在你可以执行查询，然后就可以在Mysql中看到你的查询历史和相关时间的统计信息了。如果你目前的工作对此也有需要，还等什么，快动手实现一个吧。

参考资料：
https://prestodb.github.io/docs/current/develop/event-listener.html
http://dharmeshkakadia.github.io/presto-event-listener/

欢迎关注我（WX）：叁金大数据
欢迎观看Presto免费课程：Presto实战与演练

Presto Event Listener开发

Java相关栏目本月热门文章