title: Flink Count of Events using metric
body: <p>I have a topic in kafka where i am getting multiple type of events in json format. I have created a filestreamsink to write these events to  S3 with bucketing. </p>
  
  <pre><code>FlinkKafkaConsumer errorTopicConsumer = new FlinkKafkaConsumer(ERROR_KAFKA_TOPICS,
                  new SimpleStringSchema(),
                  properties);
          final StreamingFileSink&lt;Object&gt; errorSink = StreamingFileSink
                  .forRowFormat(new Path(outputPath + "/error"), new SimpleStringEncoder&lt;&gt;("UTF-8"))
                  .withBucketAssigner(new EventTimeBucketAssignerJson())
                  .build();
  
          env.addSource(errorTopicConsumer)
                  .name("error_source")
                  .setParallelism(1)
                  .addSink(errorSink)
                  .name("error_sink").setParallelism(1);
  </code></pre>
  
  <pre><code>public class EventTimeBucketAssignerJson implements BucketAssigner&lt;Object, String&gt; {
  
      @Override
      public String getBucketId(Object record, Context context) {
          StringBuffer partitionString = new StringBuffer();
          Tuple3&lt;String, Long, String&gt; tuple3 = (Tuple3&lt;String, Long, String&gt;) record;
          try {
              partitionString.append("event_name=")
                      .append(tuple3.f0).append("/");
  
              String timePartition = TimeUtils.getEventTimeDayPartition(tuple3.f1);
              partitionString.append(timePartition);
          } catch (Exception e) {
              partitionString.append("year=").append(Constants.DEFAULT_YEAR).append("/")
                      .append("month=").append(Constants.DEFAULT_MONTH).append("/")
                      .append("day=").append(Constants.DEFAULT_DAY);
          }
          return partitionString.toString();
      }
  
      @Override
      public SimpleVersionedSerializer&lt;String&gt; getSerializer() {
          return SimpleVersionedStringSerializer.INSTANCE;
      }
  }
  </code></pre>
  
  <p>Now i want to publish hourly count of each event as metrics to prometheus and publish a grafana dashboard over that. </p>
  
  <p>So please help me how can i achieve hourly count for each event using flink metrics and publish to prometheus.</p>
  
  <p>Thanks</p>