metron-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nickal...@apache.org
Subject [29/30] metron git commit: METRON-1787 Input Time Constraints for Batch Profiler (nickwallen) closes apache/metron#1209
Date Fri, 28 Sep 2018 13:09:25 GMT
METRON-1787 Input Time Constraints for Batch Profiler (nickwallen) closes apache/metron#1209


Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/fa3be8d3
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/fa3be8d3
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/fa3be8d3

Branch: refs/heads/master
Commit: fa3be8d32ccadcd11edad046cbd063cec3a20624
Parents: 1545978
Author: nickwallen <nick@nickallen.org>
Authored: Wed Sep 26 18:13:30 2018 -0400
Committer: nickallen <nickallen@apache.org>
Committed: Wed Sep 26 18:13:30 2018 -0400

----------------------------------------------------------------------
 .../clock/EventTimeOnlyClockFactory.java        |  58 ++++++++
 .../clock/EventTimeOnlyClockFactoryTest.java    |  61 +++++++++
 .../metron-profiler-spark/README.md             |  19 +++
 .../metron/profiler/spark/BatchProfiler.java    |  38 +++++-
 .../profiler/spark/BatchProfilerConfig.java     |   6 +-
 .../metron/profiler/spark/TimestampParser.java  |  55 ++++++++
 .../spark/function/MessageRouterFunction.java   | 106 +++++++++++++--
 .../spark/BatchProfilerIntegrationTest.java     |  58 +++++++-
 .../profiler/spark/TimestampParserTest.java     |  67 ++++++++++
 .../function/MessageRouterFunctionTest.java     | 133 +++++++++++++++++--
 10 files changed, 579 insertions(+), 22 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/metron/blob/fa3be8d3/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/clock/EventTimeOnlyClockFactory.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/clock/EventTimeOnlyClockFactory.java
b/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/clock/EventTimeOnlyClockFactory.java
new file mode 100644
index 0000000..2f9ca7c
--- /dev/null
+++ b/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/clock/EventTimeOnlyClockFactory.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.metron.profiler.clock;
+
+import org.apache.metron.common.configuration.profiler.ProfilerConfig;
+
+import java.io.Serializable;
+
+/**
+ * Creates a {@link Clock} based on the profiler configuration.  This should
+ * be used in cases where only event time is accceptable.
+ *
+ * <p>If the Profiler is configured to use event time, a {@link EventTimeClock} will
+ * be created.  Otherwise, an {@link IllegalStateException} is thrown.
+ */
+public class EventTimeOnlyClockFactory implements ClockFactory, Serializable {
+
+  /**
+   * If the Profiler is configured to use event time, a {@link EventTimeClock} is created.
+   * Otherwise, an {@link IllegalArgumentException} is thrown.
+   *
+   * @param config The profiler configuration.
+   * @return The appropriate Clock based on the profiler configuration.
+   * @throws IllegalStateException If the profiler configuration is set to system time.
+   */
+  @Override
+  public Clock createClock(ProfilerConfig config) {
+    Clock clock;
+
+    boolean isEventTime = config.getTimestampField().isPresent();
+    if(isEventTime) {
+      String timestampField = config.getTimestampField().get();
+      clock = new EventTimeClock(timestampField);
+
+    } else {
+      throw new IllegalStateException("Expected profiler to use event time.");
+    }
+
+    return clock;
+  }
+}

http://git-wip-us.apache.org/repos/asf/metron/blob/fa3be8d3/metron-analytics/metron-profiler-common/src/test/java/org/apache/metron/profiler/clock/EventTimeOnlyClockFactoryTest.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-common/src/test/java/org/apache/metron/profiler/clock/EventTimeOnlyClockFactoryTest.java
b/metron-analytics/metron-profiler-common/src/test/java/org/apache/metron/profiler/clock/EventTimeOnlyClockFactoryTest.java
new file mode 100644
index 0000000..f1d4114
--- /dev/null
+++ b/metron-analytics/metron-profiler-common/src/test/java/org/apache/metron/profiler/clock/EventTimeOnlyClockFactoryTest.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.metron.profiler.clock;
+
+import org.apache.metron.common.configuration.profiler.ProfilerConfig;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Optional;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests the {@link EventTimeOnlyClockFactory}.
+ */
+public class EventTimeOnlyClockFactoryTest {
+
+  private EventTimeOnlyClockFactory clockFactory;
+
+  @Before
+  public void setup() {
+    clockFactory = new EventTimeOnlyClockFactory();
+  }
+
+  @Test
+  public void testCreateEventTimeClock() {
+    // configure the profiler to use event time
+    ProfilerConfig config = new ProfilerConfig();
+    config.setTimestampField(Optional.of("timestamp"));
+
+    // the factory should return a clock that handles 'event time'
+    Clock clock = clockFactory.createClock(config);
+    assertTrue(clock instanceof EventTimeClock);
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void testCreateProcessingTimeClock() {
+    // the profiler uses processing time by default
+    ProfilerConfig config = new ProfilerConfig();
+    clockFactory.createClock(config);
+    fail("Expected exception");
+  }
+}

http://git-wip-us.apache.org/repos/asf/metron/blob/fa3be8d3/metron-analytics/metron-profiler-spark/README.md
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-spark/README.md b/metron-analytics/metron-profiler-spark/README.md
index 99e8c7e..df143f1 100644
--- a/metron-analytics/metron-profiler-spark/README.md
+++ b/metron-analytics/metron-profiler-spark/README.md
@@ -245,6 +245,8 @@ You can store both settings for the Profiler along with settings for Spark
in th
 |---                                                                            |---
 | [`profiler.batch.input.path`](#profilerbatchinputpath)                        | The path
to the input data read by the Batch Profiler.
 | [`profiler.batch.input.format`](#profilerbatchinputformat)                    | The format
of the input data read by the Batch Profiler.
+| [`profiler.batch.input.begin`](#profilerbatchinputend)                        | Only messages
with a timestamp after this will be profiled.
+| [`profiler.batch.input.end`](#profilerbatchinputbegin)                        | Only messages
with a timestamp before this will be profiled.
 | [`profiler.period.duration`](#profilerperiodduration)                         | The duration
of each profile period.  
 | [`profiler.period.duration.units`](#profilerperioddurationunits)              | The units
used to specify the [`profiler.period.duration`](#profilerperiodduration).
 | [`profiler.hbase.salt.divisor`](#profilerhbasesaltdivisor)                    | A salt
is prepended to the row key to help prevent hot-spotting.
@@ -263,6 +265,23 @@ The path to the input data read by the Batch Profiler.
 
 The format of the input data read by the Batch Profiler.
 
+### `profiler.batch.input.begin`
+
+*Default*: undefined; no time constraint
+
+Only messages with a timestamp equal to or after this will be profiled. The Profiler will
only profiles messages with a timestamp in [`profiler.batch.input.begin`, `profiler.batch.input.end`]
inclusive.
+
+By default, no time constraint is defined. The value is expected to follow the [ISO-8601
instant format](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html#ISO_INSTANT);
2011-12-03T10:15:30Z.
+
+
+### `profiler.batch.input.end`
+
+*Default*: undefined; no time constraint
+
+Only messages with a timestamp before or equal to this will be profiled. The Profiler will
only profiles messages with a timestamp in [`profiler.batch.input.begin`, `profiler.batch.input.end`]
inclusive.
+
+By default, no time constraint is defined. The value is expected to follow the [ISO-8601
instant format](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html#ISO_INSTANT);
2011-12-03T10:15:30Z.
+
 ### `profiler.period.duration`
 
 *Default*: 15

http://git-wip-us.apache.org/repos/asf/metron/blob/fa3be8d3/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfiler.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfiler.java
b/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfiler.java
index d75abc3..39f8b3a 100644
--- a/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfiler.java
+++ b/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfiler.java
@@ -34,10 +34,12 @@ import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
 import java.lang.invoke.MethodHandles;
-import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Properties;
 
+import static org.apache.metron.profiler.spark.BatchProfilerConfig.TELEMETRY_INPUT_BEGIN;
+import static org.apache.metron.profiler.spark.BatchProfilerConfig.TELEMETRY_INPUT_END;
 import static org.apache.metron.profiler.spark.BatchProfilerConfig.TELEMETRY_INPUT_FORMAT;
 import static org.apache.metron.profiler.spark.BatchProfilerConfig.TELEMETRY_INPUT_PATH;
 import static org.apache.spark.sql.functions.sum;
@@ -51,6 +53,12 @@ public class BatchProfiler implements Serializable {
 
   protected static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
+  private TimestampParser timestampParser;
+
+  public BatchProfiler() {
+    this.timestampParser = new TimestampParser();
+  }
+
   /**
    * Execute the Batch Profiler.
    *
@@ -69,7 +77,6 @@ public class BatchProfiler implements Serializable {
 
     LOG.debug("Building {} profile(s)", profiles.getProfiles().size());
     Map<String, String> globals = Maps.fromProperties(globalProperties);
-
     String inputFormat = TELEMETRY_INPUT_FORMAT.get(profilerProps, String.class);
     String inputPath = TELEMETRY_INPUT_PATH.get(profilerProps, String.class);
     LOG.debug("Loading telemetry from '{}'", inputPath);
@@ -85,7 +92,7 @@ public class BatchProfiler implements Serializable {
 
     // find all routes for each message
     Dataset<MessageRoute> routes = telemetry
-            .flatMap(new MessageRouterFunction(profiles, globals), Encoders.bean(MessageRoute.class));
+            .flatMap(messageRouterFunction(profilerProps, profiles, globals), Encoders.bean(MessageRoute.class));
     LOG.debug("Generated {} message route(s)", routes.cache().count());
 
     // build the profiles
@@ -104,4 +111,29 @@ public class BatchProfiler implements Serializable {
 
     return count;
   }
+
+  /**
+   * Builds the function that performs message routing.
+   *
+   * @param profilerProps The profiler configuration properties.
+   * @param profiles The profile definitions.
+   * @param globals The Stellar global properties.
+   * @return A {@link MessageRouterFunction}.
+   */
+  private MessageRouterFunction messageRouterFunction(
+          Properties profilerProps,
+          ProfilerConfig profiles,
+          Map<String, String> globals) {
+    MessageRouterFunction routerFunction = new MessageRouterFunction(profiles, globals);
+
+    // an optional time constraint to limit how far back to look for telemetry
+    Optional<Long> beginAt = timestampParser.parse(TELEMETRY_INPUT_BEGIN.get(profilerProps,
String.class));
+    beginAt.ifPresent(begin -> routerFunction.withBegin(begin));
+
+    // an optional time constraint to limit the most recent telemetry
+    Optional<Long> endAt = timestampParser.parse(TELEMETRY_INPUT_END.get(profilerProps,
String.class));
+    endAt.ifPresent(end -> routerFunction.withEnd(end));
+
+    return routerFunction;
+  }
 }

http://git-wip-us.apache.org/repos/asf/metron/blob/fa3be8d3/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfilerConfig.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfilerConfig.java
b/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfilerConfig.java
index 054806e..e8cd160 100644
--- a/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfilerConfig.java
+++ b/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/BatchProfilerConfig.java
@@ -46,7 +46,11 @@ public enum BatchProfilerConfig {
 
   TELEMETRY_INPUT_FORMAT("profiler.batch.input.format", "text", String.class),
 
-  TELEMETRY_INPUT_PATH("profiler.batch.input.path", "hdfs://localhost:9000/apps/metron/indexing/indexed/*/*",
String.class);
+  TELEMETRY_INPUT_PATH("profiler.batch.input.path", "hdfs://localhost:9000/apps/metron/indexing/indexed/*/*",
String.class),
+
+  TELEMETRY_INPUT_BEGIN("profiler.batch.input.begin", "", String.class),
+
+  TELEMETRY_INPUT_END("profiler.batch.input.end", "", String.class);
 
   /**
    * The key for the configuration value.

http://git-wip-us.apache.org/repos/asf/metron/blob/fa3be8d3/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/TimestampParser.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/TimestampParser.java
b/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/TimestampParser.java
new file mode 100644
index 0000000..4be8b3e
--- /dev/null
+++ b/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/TimestampParser.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.metron.profiler.spark;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.time.Instant;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.util.Optional;
+
+/**
+ * Parses an input string and returns a timestamp in epoch milliseconds.
+ */
+public class TimestampParser {
+
+  /**
+   * Parses an input string and returns an optional timestamp in epoch milliseconds.
+   *
+   * @param inputString The input defining a timestamp.
+   * @return A timestamp in epoch milliseconds.
+   */
+  public Optional<Long> parse(String inputString) {
+    Optional<Long> epochMilli = Optional.empty();
+
+    // a blank is acceptable and treated as undefined
+    if (StringUtils.isNotBlank(inputString)) {
+      epochMilli = Optional.of(new DateTimeFormatterBuilder()
+              .append(DateTimeFormatter.ISO_INSTANT)
+              .toFormatter()
+              .parse(inputString, Instant::from)
+              .toEpochMilli());
+    }
+
+    return epochMilli;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/metron/blob/fa3be8d3/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/function/MessageRouterFunction.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/function/MessageRouterFunction.java
b/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/function/MessageRouterFunction.java
index cf8029f..31734d0 100644
--- a/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/function/MessageRouterFunction.java
+++ b/metron-analytics/metron-profiler-spark/src/main/java/org/apache/metron/profiler/spark/function/MessageRouterFunction.java
@@ -23,7 +23,11 @@ import org.apache.metron.common.configuration.profiler.ProfilerConfig;
 import org.apache.metron.profiler.DefaultMessageRouter;
 import org.apache.metron.profiler.MessageRoute;
 import org.apache.metron.profiler.MessageRouter;
+import org.apache.metron.profiler.clock.Clock;
+import org.apache.metron.profiler.clock.ClockFactory;
+import org.apache.metron.profiler.clock.EventTimeOnlyClockFactory;
 import org.apache.metron.stellar.dsl.Context;
+import org.apache.metron.stellar.dsl.StellarFunctions;
 import org.apache.spark.api.java.function.FlatMapFunction;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
@@ -54,9 +58,27 @@ public class MessageRouterFunction implements FlatMapFunction<String,
MessageRou
    */
   private ProfilerConfig profilerConfig;
 
+  /**
+   * A clock that can extract time from the messages received.
+   */
+  private Clock clock;
+
+  /**
+   * Only messages with a timestamp after this will be routed.
+   */
+  private Long begin;
+
+  /**
+   * Only messages with a timestamp before this will be routed.
+   */
+  private Long end;
+
   public MessageRouterFunction(ProfilerConfig profilerConfig, Map<String, String> globals)
{
     this.profilerConfig = profilerConfig;
     this.globals = globals;
+    this.begin = Long.MIN_VALUE;
+    this.end = Long.MAX_VALUE;
+    withClockFactory(new EventTimeOnlyClockFactory());
   }
 
   /**
@@ -70,8 +92,7 @@ public class MessageRouterFunction implements FlatMapFunction<String,
MessageRou
    */
   @Override
   public Iterator<MessageRoute> call(String jsonMessage) throws Exception {
-    List<MessageRoute> routes;
-
+    List<MessageRoute> routes = Collections.emptyList();
     JSONParser parser = new JSONParser();
     Context context = TaskUtils.getContext(globals);
     MessageRouter router = new DefaultMessageRouter(context);
@@ -80,20 +101,67 @@ public class MessageRouterFunction implements FlatMapFunction<String,
MessageRou
     Optional<JSONObject> message = toMessage(jsonMessage, parser);
     if(message.isPresent()) {
 
-      // find all routes
-      routes = router.route(message.get(), profilerConfig, context);
-      LOG.trace("Found {} route(s) for a message", routes.size());
+      // extract the timestamp from the message
+      Optional<Long> timestampOpt = clock.currentTimeMillis(message.get());
+      if (timestampOpt.isPresent()) {
+
+        // timestamp must be in [begin, end]
+        Long timestamp = timestampOpt.get();
+        if(timestamp >= begin && timestamp <= end) {
+          routes = router.route(message.get(), profilerConfig, context);
+          LOG.trace("Found {} route(s) for a message", routes.size());
+
+        } else {
+          LOG.trace("Ignoring message; timestamp={} not in [{},{}]", timestamp, prettyPrint(begin),
prettyPrint(end));
+        }
+
+      } else {
+        LOG.trace("No timestamp in message. Message will be ignored.");
+      }
 
     } else {
-      // the message is not valid and must be ignored
-      routes = Collections.emptyList();
-      LOG.trace("No route possible. Unable to parse message.");
+      LOG.trace("Unable to parse message. Message will be ignored");
     }
 
     return routes.iterator();
   }
 
   /**
+   * Set a time constraint.
+   *
+   * @param begin Only messages with a timestamp after this will be routed.
+   * @return The message router function
+   */
+  public MessageRouterFunction withBegin(Long begin) {
+    this.begin = begin;
+    return this;
+  }
+
+  /**
+   * Set a time constraint.
+   *
+   * @param end Only messages with a timestamp before this will be routed.
+   * @return The message router function
+   */
+  public MessageRouterFunction withEnd(Long end) {
+    this.end = end;
+    return this;
+  }
+
+  /**
+   * Defines the {@link ClockFactory} used to create the {@link Clock}.
+   *
+   * <p>Calling this method is only needed to override the default behavior.
+   *
+   * @param clockFactory The factory to use for creating the {@link Clock}.
+   * @return The message router function.
+   */
+  public MessageRouterFunction withClockFactory(ClockFactory clockFactory) {
+    this.clock = clockFactory.createClock(profilerConfig);
+    return this;
+  }
+
+  /**
    * Parses the raw JSON of a message.
    *
    * @param json The raw JSON to parse.
@@ -110,4 +178,26 @@ public class MessageRouterFunction implements FlatMapFunction<String,
MessageRou
       return Optional.empty();
     }
   }
+
+  /**
+   * Pretty prints a Long value for use when logging these values.
+   *
+   * <p>Long.MIN_VALUE and Long.MAX_VALUE will occur frequently and is difficult to
grok in the logs.  Instead
+   * Long.MIN_VALUE is rendered as "MIN". Long.MAX_VALUE is srendered as "MAX". All other
values are rendered
+   * directly.
+   *
+   * @param value The value to pretty print.
+   * @return
+   */
+  private static String prettyPrint(Long value) {
+    String result;
+    if(value == Long.MIN_VALUE) {
+      result = "MIN";
+    } else if(value == Long.MAX_VALUE) {
+      result = "MAX";
+    } else {
+      result = value.toString();
+    }
+    return result;
+  }
 }

http://git-wip-us.apache.org/repos/asf/metron/blob/fa3be8d3/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/BatchProfilerIntegrationTest.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/BatchProfilerIntegrationTest.java
b/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/BatchProfilerIntegrationTest.java
index 87c4246..c33644f 100644
--- a/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/BatchProfilerIntegrationTest.java
+++ b/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/BatchProfilerIntegrationTest.java
@@ -38,8 +38,11 @@ import org.junit.BeforeClass;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.lang.invoke.MethodHandles;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -51,6 +54,8 @@ import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PRO
 import static org.apache.metron.profiler.spark.BatchProfilerConfig.HBASE_COLUMN_FAMILY;
 import static org.apache.metron.profiler.spark.BatchProfilerConfig.HBASE_TABLE_NAME;
 import static org.apache.metron.profiler.spark.BatchProfilerConfig.HBASE_TABLE_PROVIDER;
+import static org.apache.metron.profiler.spark.BatchProfilerConfig.TELEMETRY_INPUT_BEGIN;
+import static org.apache.metron.profiler.spark.BatchProfilerConfig.TELEMETRY_INPUT_END;
 import static org.apache.metron.profiler.spark.BatchProfilerConfig.TELEMETRY_INPUT_FORMAT;
 import static org.apache.metron.profiler.spark.BatchProfilerConfig.TELEMETRY_INPUT_PATH;
 import static org.junit.Assert.assertTrue;
@@ -60,6 +65,7 @@ import static org.junit.Assert.assertTrue;
  */
 public class BatchProfilerIntegrationTest {
 
+  private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
   /**
    * {
    *   "timestampField": "timestamp",
@@ -212,6 +218,54 @@ public class BatchProfilerIntegrationTest {
     validateProfiles();
   }
 
+  @Test
+  public void testBatchProfilerWithEndTimeConstraint() throws Exception {
+    // the input telemetry is text/json stored in the local filesystem
+    profilerProperties.put(TELEMETRY_INPUT_PATH.getKey(), "src/test/resources/telemetry.json");
+    profilerProperties.put(TELEMETRY_INPUT_FORMAT.getKey(), "text");
+
+    // there are 40 messages before "2018-07-07T15:51:48Z" in the test data
+    profilerProperties.put(TELEMETRY_INPUT_BEGIN.getKey(), "");
+    profilerProperties.put(TELEMETRY_INPUT_END.getKey(), "2018-07-07T15:51:48Z");
+
+    BatchProfiler profiler = new BatchProfiler();
+    profiler.run(spark, profilerProperties, getGlobals(), readerProperties, getProfile());
+
+    // the max timestamp in the data is around July 7, 2018
+    assign("maxTimestamp", "1530978728982L");
+
+    // the 'window' looks up to 5 hours before the max timestamp
+    assign("window", "PROFILE_WINDOW('from 5 hours ago', maxTimestamp)");
+
+    assertTrue(execute("[12] == PROFILE_GET('count-by-ip', '192.168.66.1', window)", Boolean.class));
+    assertTrue(execute("[28] == PROFILE_GET('count-by-ip', '192.168.138.158', window)", Boolean.class));
+    assertTrue(execute("[40] == PROFILE_GET('total-count', 'total', window)", Boolean.class));
+  }
+
+  @Test
+  public void testBatchProfilerWithBeginTimeConstraint() throws Exception {
+    // the input telemetry is text/json stored in the local filesystem
+    profilerProperties.put(TELEMETRY_INPUT_PATH.getKey(), "src/test/resources/telemetry.json");
+    profilerProperties.put(TELEMETRY_INPUT_FORMAT.getKey(), "text");
+
+    // there are 60 messages after "2018-07-07T15:51:48Z" in the test data
+    profilerProperties.put(TELEMETRY_INPUT_BEGIN.getKey(), "2018-07-07T15:51:48Z");
+    profilerProperties.put(TELEMETRY_INPUT_END.getKey(), "");
+
+    BatchProfiler profiler = new BatchProfiler();
+    profiler.run(spark, profilerProperties, getGlobals(), readerProperties, getProfile());
+
+    // the max timestamp in the data is around July 7, 2018
+    assign("maxTimestamp", "1530978728982L");
+
+    // the 'window' looks up to 5 hours before the max timestamp
+    assign("window", "PROFILE_WINDOW('from 5 hours ago', maxTimestamp)");
+
+    assertTrue(execute("[14] == PROFILE_GET('count-by-ip', '192.168.66.1', window)", Boolean.class));
+    assertTrue(execute("[46] == PROFILE_GET('count-by-ip', '192.168.138.158', window)", Boolean.class));
+    assertTrue(execute("[60] == PROFILE_GET('total-count', 'total', window)", Boolean.class));
+  }
+
   /**
    * Validates the profiles that were built.
    *
@@ -263,6 +317,8 @@ public class BatchProfilerIntegrationTest {
    * @return The result of executing the Stellar expression.
    */
   private <T> T execute(String expression, Class<T> clazz) {
-    return executor.execute(expression, Collections.emptyMap(), clazz);
+    T results = executor.execute(expression, Collections.emptyMap(), clazz);
+    LOG.debug("{} = {}", expression, results);
+    return results;
   }
 }

http://git-wip-us.apache.org/repos/asf/metron/blob/fa3be8d3/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/TimestampParserTest.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/TimestampParserTest.java
b/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/TimestampParserTest.java
new file mode 100644
index 0000000..f760b35
--- /dev/null
+++ b/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/TimestampParserTest.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.metron.profiler.spark;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.time.format.DateTimeParseException;
+import java.util.Optional;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class TimestampParserTest {
+
+  private TimestampParser parser;
+
+  @Before
+  public void setup() {
+    parser = new TimestampParser();
+  }
+
+  @Test
+  public void testEmpty() {
+    Optional<Long> millis = parser.parse("");
+    assertFalse(millis.isPresent());
+  }
+
+  @Test
+  public void testBlank() {
+    Optional<Long> millis = parser.parse("      ");
+    assertFalse(millis.isPresent());
+  }
+
+  @Test
+  public void testIsoInstantFormat() {
+    // ISO-8601 instant format
+    Optional<Long> millis = parser.parse("2011-12-03T10:15:30Z");
+    assertTrue(millis.isPresent());
+    assertEquals(1322907330000L, millis.get().longValue());
+  }
+
+  @Test(expected = DateTimeParseException.class)
+  public void testInvalidFormat() {
+    parser.parse("1537502400000");
+    fail("Expected exception");
+  }
+}

http://git-wip-us.apache.org/repos/asf/metron/blob/fa3be8d3/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/function/MessageRouterFunctionTest.java
----------------------------------------------------------------------
diff --git a/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/function/MessageRouterFunctionTest.java
b/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/function/MessageRouterFunctionTest.java
index ceaa7cd..9a2cbf4 100644
--- a/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/function/MessageRouterFunctionTest.java
+++ b/metron-analytics/metron-profiler-spark/src/test/java/org/apache/metron/profiler/spark/function/MessageRouterFunctionTest.java
@@ -31,14 +31,17 @@ import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 /**
  * Tests the {@link MessageRouterFunction}.
  */
 public class MessageRouterFunctionTest {
 
+  private Long messageTimestamp = 1537468508360L;
+
   /**
-   * { "ip_src_addr": "192.168.1.22" }
+   * { "ip_src_addr": "192.168.1.22", "timestamp": 1537468508360 }
    */
   @Multiline
   private String goodMessage;
@@ -48,9 +51,15 @@ public class MessageRouterFunctionTest {
    */
   private String badMessage;
 
+  /**
+   * { "ip_src_addr": "192.168.1.22" }
+   */
+  @Multiline
+  private String messageNoTimestamp;
+
   @Test
   public void testFindRoutes() throws Exception {
-    MessageRouterFunction function = new MessageRouterFunction(oneProfile(), getGlobals());
+    MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals());
     Iterator<MessageRoute> iter = function.call(goodMessage);
 
     List<MessageRoute> routes = Lists.newArrayList(iter);
@@ -58,14 +67,20 @@ public class MessageRouterFunctionTest {
     Assert.assertEquals("profile1", routes.get(0).getProfileDefinition().getProfile());
   }
 
-  /**
-   * A bad or invalid message should return no routes.
-   */
+  @Test(expected = IllegalStateException.class)
+  public void testWithSystemTime() throws Exception {
+    MessageRouterFunction function = new MessageRouterFunction(profileWithSystemTime(), getGlobals());
+    Iterator<MessageRoute> iter = function.call(goodMessage);
+
+    Assert.fail("Exception expected as system time is not supported.");
+  }
+
   @Test
   public void testWithBadMessage() throws Exception {
-    MessageRouterFunction function = new MessageRouterFunction(oneProfile(), getGlobals());
+    MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals());
     Iterator<MessageRoute> iter = function.call(badMessage);
 
+    // an invalid message should return no routes
     List<MessageRoute> routes = Lists.newArrayList(iter);
     Assert.assertEquals(0, routes.size());
   }
@@ -81,7 +96,88 @@ public class MessageRouterFunctionTest {
     Assert.assertEquals("profile2", routes.get(1).getProfileDefinition().getProfile());
   }
 
-  private ProfilerConfig oneProfile() {
+  @Test
+  public void testWithNoTimestampInMessage() throws Exception {
+    MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals());
+    Iterator<MessageRoute> iter = function.call(messageNoTimestamp);
+
+    // with no timestamp, the message should be ignored
+    List<MessageRoute> routes = Lists.newArrayList(iter);
+    Assert.assertEquals(0, routes.size());
+  }
+
+  @Test
+  public void testMessageFilteredByBegin() throws Exception {
+    MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals())
+            .withBegin(messageTimestamp + 1000);
+    Iterator<MessageRoute> iter = function.call(goodMessage);
+
+    // the message should be filtered because it is before `beginAt`
+    List<MessageRoute> routes = Lists.newArrayList(iter);
+    Assert.assertEquals(0, routes.size());
+  }
+
+  @Test
+  public void testMessageNotFilteredByBegin() throws Exception {
+    MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals())
+            .withBegin(messageTimestamp - 1000);
+    Iterator<MessageRoute> iter = function.call(goodMessage);
+
+    // the message should NOT be filtered because it is after 'beginAt'
+    List<MessageRoute> routes = Lists.newArrayList(iter);
+    Assert.assertEquals(1, routes.size());
+  }
+
+  @Test
+  public void testMessageFilteredByEnd() throws Exception {
+    MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals())
+            .withEnd(messageTimestamp - 1000);
+    Iterator<MessageRoute> iter = function.call(goodMessage);
+
+    // the message should be filtered because it is after 'endAt'
+    List<MessageRoute> routes = Lists.newArrayList(iter);
+    Assert.assertEquals(0, routes.size());
+  }
+
+  @Test
+  public void testMessageNotFilteredByEnd() throws Exception {
+    MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals())
+            .withEnd(messageTimestamp + 1000);
+    Iterator<MessageRoute> iter = function.call(goodMessage);
+
+    // the message should NOT be filtered because it is before 'endAt'
+    List<MessageRoute> routes = Lists.newArrayList(iter);
+    Assert.assertEquals(1, routes.size());
+  }
+
+  @Test
+  public void testMessageFilteredByBeginAndEnd() throws Exception {
+    MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals())
+            .withBegin(messageTimestamp + 1000)
+            .withEnd(messageTimestamp + 2000);
+    Iterator<MessageRoute> iter = function.call(goodMessage);
+
+    // the message should be filtered because it is outside of [beginAt, endAt]
+    List<MessageRoute> routes = Lists.newArrayList(iter);
+    Assert.assertEquals(0, routes.size());
+  }
+
+  @Test
+  public void testMessageNotFilteredByBeginAndEnd() throws Exception {
+    MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals())
+            .withBegin(messageTimestamp - 1000)
+            .withEnd(messageTimestamp + 1000);
+    Iterator<MessageRoute> iter = function.call(goodMessage);
+
+    // the message should NOT be filtered because it is after 'endAt'
+    List<MessageRoute> routes = Lists.newArrayList(iter);
+    Assert.assertEquals(1, routes.size());
+  }
+
+  /**
+   * Creates a profiler definition, using event time, and containing one profile.
+   */
+  private ProfilerConfig profile() {
     ProfileConfig profile = new ProfileConfig()
             .withProfile("profile1")
             .withForeach("ip_src_addr")
@@ -89,9 +185,13 @@ public class MessageRouterFunctionTest {
             .withResult("count");
 
     return new ProfilerConfig()
-            .withProfile(profile);
+            .withProfile(profile)
+            .withTimestampField(Optional.of("timestamp"));
   }
 
+  /**
+   * Creates a profiler definition, using event time, and containing two profiles.
+   */
   private ProfilerConfig twoProfiles() {
     ProfileConfig profile1 = new ProfileConfig()
             .withProfile("profile1")
@@ -105,7 +205,22 @@ public class MessageRouterFunctionTest {
             .withResult("count");
     return new ProfilerConfig()
             .withProfile(profile1)
-            .withProfile(profile2);
+            .withProfile(profile2)
+            .withTimestampField(Optional.of("timestamp"));
+  }
+
+  /**
+   * Creates a profiler definition using system time.
+   */
+  private ProfilerConfig profileWithSystemTime() {
+    ProfileConfig profile = new ProfileConfig()
+            .withProfile("profile1")
+            .withForeach("ip_src_addr")
+            .withUpdate("count", "count + 1")
+            .withResult("count");
+
+    return new ProfilerConfig()
+            .withProfile(profile);
   }
 
   private Map<String, String> getGlobals() {


Mime
View raw message