/** * Get the value which %percent% of the values are less than. This works * just like median (where median represents the 50% point). A typical * desire is to see the 90% point - the value that 90% of the data points * are below, the remaining 10% are above. * * @param percent * number representing the wished percent (between <code>0</code> * and <code>1.0</code>) * @return the value which %percent% of the values are less than */ public T getPercentPoint(double percent) { if (count <= 0) { return ZERO; } if (percent >= 1.0) { return getMax(); } // use Math.round () instead of simple (long) to provide correct value rounding long target = Math.round (count * percent); try { for (Entry<T, MutableLong> val : valuesMap.entrySet()) { target -= val.getValue().longValue(); if (target <= 0){ return val.getKey(); } } } catch (ConcurrentModificationException ignored) { // ignored. May happen occasionally, but no harm done if so. } return ZERO; // TODO should this be getMin()? }
@Override public <STREAM extends WindowedStream<Tuple.WindowedTuple<Long>>> STREAM count(Option... opts) { Function.MapFunction<T, Tuple<Long>> kVMap = new Function.MapFunction<T, Tuple<Long>>() { @Override public Tuple<Long> f(T input) { if (input instanceof Tuple.TimestampedTuple) { return new Tuple.TimestampedTuple<>(((Tuple.TimestampedTuple)input).getTimestamp(), 1L); } else { return new Tuple.TimestampedTuple<>(System.currentTimeMillis(), 1L); } } }; WindowedStream<Tuple<Long>> innerstream = map(kVMap); WindowedOperatorImpl<Long, MutableLong, Long> windowedOperator = createWindowedOperator(new SumLong()); return innerstream.addOperator(windowedOperator, windowedOperator.input, windowedOperator.output, opts); }
void bucketAccessed(long bucketId) { long now = System.currentTimeMillis(); if (accessedBucketIds.add(bucketId) || now - lastUpdateAccessTime > updateAccessTimeInterval) { synchronized (bucketLastAccess) { for (long id : accessedBucketIds) { MutableLong lastAccessTime = bucketLastAccess.get(id); if (lastAccessTime != null) { lastAccessTime.setValue(now); } else { bucketLastAccess.put(id, new MutableLong(now)); } } } accessedBucketIds.clear(); lastUpdateAccessTime = now; } }
@Test public void testSimpleRemoveEmpty() { WindowEndQueueManager<Query, Void> wqqm = new WindowEndQueueManager<>(); wqqm.setup(null); wqqm.beginWindow(0); QueryBundle<Query, Void, MutableLong> qb = wqqm.dequeue(); Query queryD = qb == null ? null : qb.getQuery(); Assert.assertEquals("The queries must match.", null, queryD); qb = wqqm.dequeue(); queryD = qb == null ? null : qb.getQuery(); Assert.assertEquals("The queries must match.", null, queryD); wqqm.endWindow(); wqqm.teardown(); }
@Test public void testSimpleAddOneRemove() { WindowEndQueueManager<Query, Void> wqqm = new WindowEndQueueManager<>(); wqqm.setup(null); wqqm.beginWindow(0); Query query = new MockQuery("1"); wqqm.enqueue(query, null, new MutableLong(1L)); Query queryD = wqqm.dequeue().getQuery(); QueryBundle<Query, Void, MutableLong> qb = wqqm.dequeue(); Query queryD1 = qb == null ? null : qb.getQuery(); wqqm.endWindow(); wqqm.teardown(); Assert.assertEquals("The queries must match.", query, queryD); Assert.assertEquals("The queries must match.", null, queryD1); }
@Override public void run() { int numLoops = totalTuples / batchSize; for (int loopCounter = 0, tupleCounter = 0; loopCounter < numLoops; loopCounter++, tupleCounter++) { for (int batchCounter = 0; batchCounter < batchSize; batchCounter++, tupleCounter++) { queueManager.enqueue(new MockQuery(tupleCounter + ""), null, new MutableLong(1L)); if (rand.nextDouble() < waitMillisProb) { try { Thread.sleep(1); } catch (InterruptedException ex) { throw new RuntimeException(ex); } } } } }
private WindowedOperatorImpl<Long, MutableLong, Long> createDefaultWindowedOperator() { WindowedOperatorImpl<Long, MutableLong, Long> windowedOperator = new WindowedOperatorImpl<>(); if (useSpillable) { sccImpl = new SpillableComplexComponentImpl(testMeta.timeStore); // TODO: We don't yet support Spillable data structures for window state storage because SpillableMapImpl does not yet support iterating over all keys. windowStateStorage = new InMemoryWindowedStorage<>(); SpillableWindowedPlainStorage<MutableLong> pds = new SpillableWindowedPlainStorage<>(); pds.setSpillableComplexComponent(sccImpl); plainDataStorage = pds; SpillableWindowedPlainStorage<Long> prs = new SpillableWindowedPlainStorage<>(); prs.setSpillableComplexComponent(sccImpl); plainRetractionStorage = prs; windowedOperator.addComponent("SpillableComplexComponent", sccImpl); } else { windowStateStorage = new InMemoryWindowedStorage<>(); plainDataStorage = new InMemoryWindowedStorage<>(); plainRetractionStorage = new InMemoryWindowedStorage<>(); } windowedOperator.setDataStorage(plainDataStorage); windowedOperator.setRetractionStorage(plainRetractionStorage); windowedOperator.setWindowStateStorage(windowStateStorage); windowedOperator.setAccumulation(new SumAccumulation()); return windowedOperator; }
@Test public void testValidation() throws Exception { WindowedOperatorImpl<Long, MutableLong, Long> windowedOperator = new WindowedOperatorImpl<>(); verifyValidationFailure(windowedOperator, "nothing is configured"); windowedOperator.setWindowStateStorage(new InMemoryWindowedStorage<WindowState>()); verifyValidationFailure(windowedOperator, "data storage is not set"); windowedOperator.setDataStorage(new InMemoryWindowedStorage<MutableLong>()); verifyValidationFailure(windowedOperator, "accumulation is not set"); windowedOperator.setAccumulation(new SumAccumulation()); windowedOperator.validate(); windowedOperator.setTriggerOption(new TriggerOption().accumulatingAndRetractingFiredPanes()); verifyValidationFailure(windowedOperator, "retracting storage is not set for ACCUMULATING_AND_RETRACTING"); windowedOperator.setRetractionStorage(new InMemoryWindowedStorage<Long>()); windowedOperator.validate(); windowedOperator.setTriggerOption(new TriggerOption().discardingFiredPanes().firingOnlyUpdatedPanes()); verifyValidationFailure(windowedOperator, "DISCARDING is not valid for option firingOnlyUpdatedPanes"); windowedOperator.setTriggerOption(new TriggerOption().accumulatingFiredPanes().firingOnlyUpdatedPanes()); windowedOperator.setRetractionStorage(null); verifyValidationFailure(windowedOperator, "retracting storage is not set for option firingOnlyUpdatedPanes"); }
@Test public void testSlidingWindowAssignment() { WindowedOperatorImpl<Long, MutableLong, Long> windowedOperator = createDefaultWindowedOperator(); windowedOperator.setWindowOption(new WindowOption.SlidingTimeWindows(Duration.millis(1000), Duration.millis(200))); windowedOperator.setup(testMeta.operatorContext); Tuple.WindowedTuple<Long> windowedValue = windowedOperator.getWindowedValue(new Tuple.TimestampedTuple<>(BASE + 1600L, 2L)); Collection<? extends Window> windows = windowedValue.getWindows(); Window[] winArray = windows.toArray(new Window[]{}); Assert.assertEquals(5, winArray.length); Assert.assertEquals(BASE + 800, winArray[0].getBeginTimestamp()); Assert.assertEquals(1000, winArray[0].getDurationMillis()); Assert.assertEquals(BASE + 1000, winArray[1].getBeginTimestamp()); Assert.assertEquals(1000, winArray[1].getDurationMillis()); Assert.assertEquals(BASE + 1200, winArray[2].getBeginTimestamp()); Assert.assertEquals(1000, winArray[2].getDurationMillis()); Assert.assertEquals(BASE + 1400, winArray[3].getBeginTimestamp()); Assert.assertEquals(1000, winArray[3].getDurationMillis()); Assert.assertEquals(BASE + 1600, winArray[4].getBeginTimestamp()); Assert.assertEquals(1000, winArray[4].getDurationMillis()); windowedOperator.teardown(); }
@Test public void testKeyedAccumulation() { KeyedWindowedOperatorImpl<String, Long, MutableLong, Long> windowedOperator = createDefaultKeyedWindowedOperator(false); windowedOperator.setWindowOption(new WindowOption.TimeWindows(Duration.millis(1000))); windowedOperator.setup(testMeta.operatorContext); windowedOperator.beginWindow(1); windowedOperator.processTuple(new Tuple.TimestampedTuple<>(BASE + 100L, new KeyValPair<>("a", 2L))); windowedOperator.processTuple(new Tuple.TimestampedTuple<>(BASE + 200L, new KeyValPair<>("a", 3L))); windowedOperator.processTuple(new Tuple.TimestampedTuple<>(BASE + 300L, new KeyValPair<>("b", 4L))); windowedOperator.processTuple(new Tuple.TimestampedTuple<>(BASE + 150L, new KeyValPair<>("b", 5L))); windowedOperator.endWindow(); Assert.assertEquals(1, keyedDataStorage.size()); Assert.assertEquals(5L, keyedDataStorage.get(new Window.TimeWindow(BASE, 1000), "a").longValue()); Assert.assertEquals(9L, keyedDataStorage.get(new Window.TimeWindow(BASE, 1000), "b").longValue()); windowedOperator.teardown(); }
@Override public void populateDAG(DAG dag, Configuration configuration) { WordGenerator inputOperator = new WordGenerator(); KeyedWindowedOperatorImpl<String, Long, MutableLong, Long> windowedOperator = new KeyedWindowedOperatorImpl<>(); Accumulation<Long, MutableLong, Long> sum = new SumAccumulation(); windowedOperator.setAccumulation(sum); windowedOperator.setDataStorage(new InMemoryWindowedKeyedStorage<String, MutableLong>()); windowedOperator.setRetractionStorage(new InMemoryWindowedKeyedStorage<String, Long>()); windowedOperator.setWindowStateStorage(new InMemoryWindowedStorage<WindowState>()); windowedOperator.setWindowOption(new WindowOption.TimeWindows(Duration.standardMinutes(1))); windowedOperator.setTriggerOption(TriggerOption.AtWatermark().withEarlyFiringsAtEvery(Duration.millis(1000)).accumulatingAndRetractingFiredPanes()); //windowedOperator.setAllowedLateness(Duration.millis(14000)); ConsoleOutputOperator outputOperator = new ConsoleOutputOperator(); dag.addOperator("inputOperator", inputOperator); dag.addOperator("windowedOperator", windowedOperator); dag.addOperator("outputOperator", outputOperator); dag.addStream("input_windowed", inputOperator.output, windowedOperator.input); dag.addStream("windowed_output", windowedOperator.output, outputOperator.input); }
@Override public void populateDAG(DAG dag, Configuration configuration) { RandomNumberPairGenerator inputOperator = new RandomNumberPairGenerator(); WindowedOperatorImpl<MutablePair<Double, Double>, MutablePair<MutableLong, MutableLong>, Double> windowedOperator = new WindowedOperatorImpl<>(); Accumulation<MutablePair<Double, Double>, MutablePair<MutableLong, MutableLong>, Double> piAccumulation = new PiAccumulation(); windowedOperator.setAccumulation(piAccumulation); windowedOperator.setDataStorage(new InMemoryWindowedStorage<MutablePair<MutableLong, MutableLong>>()); windowedOperator.setWindowStateStorage(new InMemoryWindowedStorage<WindowState>()); windowedOperator.setWindowOption(new WindowOption.GlobalWindow()); windowedOperator.setTriggerOption(TriggerOption.AtWatermark().withEarlyFiringsAtEvery(Duration.millis(1000)).accumulatingFiredPanes()); ConsoleOutputOperator outputOperator = new ConsoleOutputOperator(); dag.addOperator("inputOperator", inputOperator); dag.addOperator("windowedOperator", windowedOperator); dag.addOperator("outputOperator", outputOperator); dag.addStream("input_windowed", inputOperator.output, windowedOperator.input); dag.addStream("windowed_output", windowedOperator.output, outputOperator.input); }
@Test public void SumTest() { SumInt si = new SumInt(); SumLong sl = new SumLong(); SumFloat sf = new SumFloat(); SumDouble sd = new SumDouble(); Assert.assertEquals(new MutableInt(10), si.accumulate(si.defaultAccumulatedValue(), 10)); Assert.assertEquals(new MutableInt(11), si.accumulate(new MutableInt(1), 10)); Assert.assertEquals(new MutableInt(22), si.merge(new MutableInt(1), new MutableInt(21))); Assert.assertEquals(new MutableLong(10L), sl.accumulate(sl.defaultAccumulatedValue(), 10L)); Assert.assertEquals(new MutableLong(22L), sl.accumulate(new MutableLong(2L), 20L)); Assert.assertEquals(new MutableLong(41L), sl.merge(new MutableLong(32L), new MutableLong(9L))); Assert.assertEquals(new MutableFloat(9.0F), sf.accumulate(sf.defaultAccumulatedValue(), 9.0F)); Assert.assertEquals(new MutableFloat(22.5F), sf.accumulate(new MutableFloat(2.5F), 20F)); Assert.assertEquals(new MutableFloat(41.0F), sf.merge(new MutableFloat(33.1F), new MutableFloat(7.9F))); Assert.assertEquals(new MutableDouble(9.0), sd.accumulate(sd.defaultAccumulatedValue(), 9.0)); Assert.assertEquals(new MutableDouble(22.5), sd.accumulate(new MutableDouble(2.5), 20.0)); Assert.assertEquals(new MutableDouble(41.0), sd.merge(new MutableDouble(33.1), new MutableDouble(7.9))); }
/** * Compute gauss mask 0. * * @param num * the num * @param sigma * the sigma * @return the double[] */ /* * num ist eigentlich pointer - aufrufende Funkion nimmt an, dass num geändert * wird. Übergebe es deswegen als MutableDouble aus CommonsLang */ public double[] compute_gauss_mask_0(MutableLong num, double sigma) { int i, n; double limit; double[] h; limit = LinesUtil.MASK_SIZE(LinesUtil.MAX_SIZE_MASK_0, sigma); /* Error < 0.001 on each side */ n = (int) limit; h = new double[2 * n + 1]; for (i = -n + 1; i <= n - 1; i++) h[n + i] = phi0(-i + 0.5, sigma) - phi0(-i - 0.5, sigma); h[0] = 1.0 - phi0(n - 0.5, sigma); h[2 * n] = phi0(-n + 0.5, sigma); num.setValue(n); return h; }
/** * Compute gauss mask 1. * * @param num * the num * @param sigma * the sigma * @return the double[] */ /* * num ist eigentlich pointer - aufrufende Funkion nimmt an, dass num geändert * wird. Übergebe es deswegen als MutableDouble aus CommonsLang */ public double[] compute_gauss_mask_1(MutableLong num, double sigma) { int i, n; double limit; double[] h; limit = LinesUtil.MASK_SIZE(LinesUtil.MAX_SIZE_MASK_1, sigma); /* Error < 0.001 on each side */ n = (int) limit; h = new double[2 * n + 1]; for (i = -n + 1; i <= n - 1; i++) h[n + i] = phi1(-i + 0.5, sigma) - phi1(-i - 0.5, sigma); h[0] = -phi1(n - 0.5, sigma); h[2 * n] = phi1(-n + 0.5, sigma); num.setValue(n); return h; }
/** * Compute gauss mask 2. * * @param num * the num * @param sigma * the sigma * @return the double[] */ /* * num ist eigentlich pointer - aufrufende Funkion nimmt an, dass num geändert * wird. Übergebe es deswegen als MutableDouble aus CommonsLang */ public double[] compute_gauss_mask_2(MutableLong num, double sigma) { int i, n; double limit; double[] h; limit = LinesUtil.MASK_SIZE(LinesUtil.MAX_SIZE_MASK_2, sigma); /* Error < 0.001 on each side */ n = (int) limit; h = new double[2 * n + 1]; for (i = -n + 1; i <= n - 1; i++) h[n + i] = phi2(-i + 0.5, sigma) - phi2(-i - 0.5, sigma); h[0] = -phi2(n - 0.5, sigma); h[2 * n] = phi2(-n + 0.5, sigma); num.setValue(n); return h; }
protected final long stampSequenceIdAndPublishToRingBuffer(RegionInfo hri, WALKeyImpl key, WALEdit edits, boolean inMemstore, RingBuffer<RingBufferTruck> ringBuffer) throws IOException { if (this.closed) { throw new IOException( "Cannot append; log is closed, regionName = " + hri.getRegionNameAsString()); } MutableLong txidHolder = new MutableLong(); MultiVersionConcurrencyControl.WriteEntry we = key.getMvcc().begin(() -> { txidHolder.setValue(ringBuffer.next()); }); long txid = txidHolder.longValue(); try (TraceScope scope = TraceUtil.createTrace(implClassName + ".append")) { FSWALEntry entry = new FSWALEntry(txid, key, edits, hri, inMemstore); entry.stampRegionSequenceId(we); ringBuffer.get(txid).load(entry); } finally { ringBuffer.publish(txid); } return txid; }
@Override public long freeMemory() { MutableLong totalBytesReleased = new MutableLong(0); ifNotClosed(() -> { for (FileBucket bucket : fileBuckets) { bucket.lockRead(); for (FileInfo fileInfo : bucket.getFiles()) { long bytesReleased = fileInfo.discardFileContents(); updateSizeOfCachedFileContents(-bytesReleased); totalBytesReleased.add(bytesReleased); } bucket.unlockRead(); } }); return totalBytesReleased.longValue(); }
/** * Get the value which %percent% of the values are less than. This works * just like median (where median represents the 50% point). A typical * desire is to see the 90% point - the value that 90% of the data points * are below, the remaining 10% are above. * * @param percent * @return the value which %percent% of the values are less than */ public T getPercentPoint(double percent) { if (count <= 0) { return ZERO; } if (percent >= 1.0) { return getMax(); } // use Math.round () instead of simple (long) to provide correct value rounding long target = Math.round (count * percent); try { for (Entry<T, MutableLong> val : valuesMap.entrySet()) { target -= val.getValue().longValue(); if (target <= 0){ return val.getKey(); } } } catch (ConcurrentModificationException ignored) { // ignored. May happen occasionally, but no harm done if so. } return ZERO; // TODO should this be getMin()? }
@Override public void add(String value) { if (values.containsKey(value)) { values.get(value).increment(); } else { values.put(value, new MutableLong(1)); } }
@Override public void merge(Accumulator<String, ConcurrentSkipListMap<String, MutableLong>> other) { other.getLocalValue().forEach((k, v) -> { if (values.containsKey(k)) { values.get(k).add(v.longValue()); } else { values.put(k, v); } }); }
/** * <p>Default constructor. Creates an empty design</p> * * @since 0.4.0 */ public NetPlan() { super(null, 0, 0, new AttributeMap()); this.netPlan = this; DEFAULT_ROUTING_TYPE = RoutingType.SOURCE_ROUTING; isModifiable = true; networkDescription = ""; networkName = ""; nextElementId = new MutableLong(1); layers = new ArrayList<NetworkLayer>(); nodes = new ArrayList<Node>(); srgs = new ArrayList<SharedRiskGroup>(); resources = new ArrayList<Resource>(); cache_nodesDown = new HashSet<Node>(); this.cache_type2Resources = new HashMap<String, Set<Resource>>(); this.cache_id2NodeMap = new HashMap<Long, Node>(); this.cache_id2ResourceMap = new HashMap<Long, Resource>(); this.cache_id2LayerMap = new HashMap<Long, NetworkLayer>(); this.cache_id2srgMap = new HashMap<Long, SharedRiskGroup>(); this.cache_id2LinkMap = new HashMap<Long, Link>(); this.cache_id2DemandMap = new HashMap<Long, Demand>(); this.cache_id2MulticastDemandMap = new HashMap<Long, MulticastDemand>(); this.cache_id2RouteMap = new HashMap<Long, Route>(); this.cache_id2MulticastTreeMap = new HashMap<Long, MulticastTree>(); this.cache_taggedElements = new HashMap<> (); this.cache_nodesPerSiteName = new HashMap<> (); this.cache_planningDomain2nodes = new HashMap<> (); interLayerCoupling = new DirectedAcyclicGraph<NetworkLayer, DemandLinkMapping>(DemandLinkMapping.class); defaultLayer = addLayer("Layer 0", null, null, null, null, null); }
/** * Returns the distribution of the values in the list. * * @return map containing either Integer or Long keys; entries are a Number array containing the key and the [Integer] count. * TODO - why is the key value also stored in the entry array? See Bug 53825 */ public Map<Number, Number[]> getDistribution() { Map<Number, Number[]> items = new HashMap<>(); for (Entry<T, MutableLong> entry : valuesMap.entrySet()) { Number[] dis = new Number[2]; dis[0] = entry.getKey(); dis[1] = entry.getValue(); items.put(entry.getKey(), dis); } return items; }
private void updateValueCount(T actualValue, long sampleCount) { MutableLong count = valuesMap.get(actualValue); if (count != null) { count.add(sampleCount); } else { // insert new value valuesMap.put(actualValue, new MutableLong(sampleCount)); } }
@Override public <K, STREAM extends WindowedStream<Tuple.WindowedTuple<KeyValPair<K, Long>>>> STREAM countByKey(Function.ToKeyValue<T, K, Long> convertToKeyValue, Option... opts) { WindowedStream<Tuple<KeyValPair<K, Long>>> kvstream = map(convertToKeyValue); KeyedWindowedOperatorImpl<K, Long, MutableLong, Long> keyedWindowedOperator = createKeyedWindowedOperator(new SumLong()); return kvstream.addOperator(keyedWindowedOperator, keyedWindowedOperator.input, keyedWindowedOperator.output, opts); }
@Override public void endWindow() { for (QueueListNode<QueryBundle<QUERY_TYPE, META_QUERY, MutableLong>> tempNode = queryQueue.getHead(); tempNode != null; tempNode = tempNode.getNext()) { MutableLong qc = tempNode.getPayload().getQueueContext(); qc.decrement(); } }
@Override public boolean enqueue(QUERY query, META_QUERY metaQuery, MutableLong context) { if (context != null) { query.setCountdown(context.getValue()); } if (query.isOneTime()) { return super.enqueue(query, metaQuery, new MutableLong(1L)); } else { return super.enqueue(query, metaQuery, new MutableLong(query.getCountdown())); } }
@Override public boolean addingFilter(QueryBundle<QUERY, META_QUERY, MutableLong> queryBundle) { QueueListNode<QueryBundle<QUERY, META_QUERY, MutableLong>> queryNode = queryIDToNode.get(queryBundle.getQuery().getId()); if (queryNode == null) { return true; } queryNode.setPayload(queryBundle); return false; }
@Override public Result executeQuery(Query query, Void metaQuery, MutableLong queueContext) { return new DataResultSnapshot(query, currentData, queueContext.getValue()); }
@Test public void testSimpleAddRemove2() { WindowEndQueueManager<Query, Void> wqqm = new WindowEndQueueManager<>(); wqqm.setup(null); wqqm.beginWindow(0); Query query = new MockQuery("1"); wqqm.enqueue(query, null, new MutableLong(1L)); Query queryD = wqqm.dequeue().getQuery(); QueryBundle<Query, Void, MutableLong> qb = wqqm.dequeue(); Query queryD1 = qb == null ? null : qb.getQuery(); Query query1 = new MockQuery("2"); wqqm.enqueue(query1, null, new MutableLong(1L)); Query query1D = wqqm.dequeue().getQuery(); qb = wqqm.dequeue(); Query query1D1 = qb == null ? null : qb.getQuery(); wqqm.endWindow(); wqqm.teardown(); Assert.assertEquals("The queries must match.", query, queryD); Assert.assertEquals("The queries must match.", null, queryD1); Assert.assertEquals("The queries must match.", query1, query1D); Assert.assertEquals("The queries must match.", null, query1D1); }
@Test public void testSimpleAddAfterStarted() { WindowEndQueueManager<Query, Void> wqqm = new WindowEndQueueManager<>(); wqqm.setup(null); wqqm.beginWindow(0); Query query = new MockQuery("0"); wqqm.enqueue(query, null, new MutableLong(1L)); Query query1 = new MockQuery("1"); wqqm.enqueue(query1, null, new MutableLong(1L)); Query queryD = wqqm.dequeue().getQuery(); Query query2 = new MockQuery("2"); wqqm.enqueue(query2, null, new MutableLong(1L)); Query query1D = wqqm.dequeue().getQuery(); Query query2D = wqqm.dequeue().getQuery(); QueryBundle<Query, Void, MutableLong> qb = wqqm.dequeue(); Query query3D = qb == null ? null : qb.getQuery(); wqqm.endWindow(); wqqm.teardown(); Assert.assertEquals("The queries must match.", query, queryD); Assert.assertEquals("The queries must match.", query1, query1D); Assert.assertEquals("The queries must match.", query2, query2D); Assert.assertEquals("The queries must match.", null, query3D); }
@Override public MockResult executeQuery(MockQuery query, Void metaQuery, MutableLong queueContext) { if (rand.nextDouble() < waitMillisProb) { try { Thread.sleep(1); } catch (InterruptedException ex) { throw new RuntimeException(ex); } } return new MockResult(query); }
@Test public void testTimeBucketKeyExpiry() { final MutableLong purgeLessThanEqualTo = new MutableLong(-2); testMeta.timeBucketAssigner.setExpireBefore(Duration.standardSeconds(1)); testMeta.timeBucketAssigner.setBucketSpan(Duration.standardSeconds(1)); testMeta.timeBucketAssigner.setPurgeListener(new TimeBucketAssigner.PurgeListener() { @Override public void purgeTimeBucketsLessThanEqualTo(long timeBucket) { purgeLessThanEqualTo.setValue(timeBucket); } }); long referenceTime = testMeta.timeBucketAssigner.getReferenceInstant().getMillis(); testMeta.timeBucketAssigner.setup(testMeta.mockManagedStateContext); Assert.assertEquals("purgeLessThanEqualTo", -2L, purgeLessThanEqualTo.longValue()); long time0 = Duration.standardSeconds(0).getMillis() + referenceTime; Assert.assertEquals("time bucket", 1, testMeta.timeBucketAssigner.getTimeBucket(time0) ); testMeta.timeBucketAssigner.endWindow(); Assert.assertEquals("purgeLessThanEqualTo", -1, purgeLessThanEqualTo.longValue()); long time1 = Duration.standardSeconds(9).getMillis() + referenceTime; Assert.assertEquals("time bucket", 10, testMeta.timeBucketAssigner.getTimeBucket(time1) ); testMeta.timeBucketAssigner.endWindow(); Assert.assertEquals("purgeLessThanEqualTo", 8, purgeLessThanEqualTo.longValue()); long time2 = Duration.standardSeconds(10).getMillis() + referenceTime; Assert.assertEquals("time bucket", 11, testMeta.timeBucketAssigner.getTimeBucket(time2) ); testMeta.timeBucketAssigner.endWindow(); Assert.assertEquals("purgeLessThanEqualTo", 9, purgeLessThanEqualTo.longValue()); //Check for expiry of time1 now Assert.assertEquals("time bucket", -1, testMeta.timeBucketAssigner.getTimeBucket(time1) ); testMeta.timeBucketAssigner.endWindow(); Assert.assertEquals("purgeLessThanEqualTo", 9, purgeLessThanEqualTo.longValue()); testMeta.timeBucketAssigner.teardown(); }
private KeyedWindowedOperatorImpl<String, Long, MutableLong, Long> createDefaultKeyedWindowedOperator(boolean forSession) { KeyedWindowedOperatorImpl<String, Long, MutableLong, Long> windowedOperator = new KeyedWindowedOperatorImpl<>(); if (useSpillable) { sccImpl = new SpillableComplexComponentImpl(testMeta.timeStore); // TODO: We don't yet support Spillable data structures for window state storage because SpillableMapImpl does not yet support iterating over all keys. windowStateStorage = new InMemoryWindowedStorage<>(); if (forSession) { SpillableSessionWindowedStorage<String, MutableLong> sws = new SpillableSessionWindowedStorage<>(); sws.setSpillableComplexComponent(sccImpl); keyedDataStorage = sws; } else { SpillableWindowedKeyedStorage<String, MutableLong> kds = new SpillableWindowedKeyedStorage<>(); kds.setSpillableComplexComponent(sccImpl); keyedDataStorage = kds; } SpillableWindowedKeyedStorage<String, Long> krs = new SpillableWindowedKeyedStorage<>(); krs.setSpillableComplexComponent(sccImpl); keyedRetractionStorage = krs; windowedOperator.addComponent("SpillableComplexComponent", sccImpl); } else { windowStateStorage = new InMemoryWindowedStorage<>(); if (forSession) { keyedDataStorage = new InMemorySessionWindowedStorage<>(); } else { keyedDataStorage = new InMemoryWindowedKeyedStorage<>(); } keyedRetractionStorage = new InMemoryWindowedKeyedStorage<>(); } windowedOperator.setDataStorage(keyedDataStorage); windowedOperator.setRetractionStorage(keyedRetractionStorage); windowedOperator.setWindowStateStorage(windowStateStorage); windowedOperator.setAccumulation(new SumAccumulation()); return windowedOperator; }
@Test public void testImplicitWatermarks() { WindowedOperatorImpl<Long, MutableLong, Long> windowedOperator = createDefaultWindowedOperator(); CollectorTestSink controlSink = new CollectorTestSink(); windowedOperator.controlOutput.setSink(controlSink); windowedOperator.setWindowOption(new WindowOption.TimeWindows(Duration.millis(1000))); windowedOperator.setAllowedLateness(Duration.millis(1000)); windowedOperator.setImplicitWatermarkGenerator(new FixedDiffEventTimeWatermarkGen(100)); windowedOperator.setup(testMeta.operatorContext); windowedOperator.beginWindow(1); windowedOperator.endWindow(); Assert.assertEquals("We should get no watermark tuple", 0, controlSink.getCount(false)); windowedOperator.beginWindow(2); windowedOperator.processTuple(new Tuple.TimestampedTuple<>(BASE + 100L, 2L)); windowedOperator.endWindow(); Assert.assertEquals("We should get one watermark tuple", 1, controlSink.getCount(false)); Assert.assertEquals("Check Watermark value", ((ControlTuple.Watermark)controlSink.collectedTuples.get(0)).getTimestamp(), BASE); windowedOperator.beginWindow(3); windowedOperator.processTuple(new Tuple.TimestampedTuple<>(BASE + 900L, 4L)); windowedOperator.endWindow(); Assert.assertEquals("We should get two watermark tuples", 2, controlSink.getCount(false)); Assert.assertEquals("Check Watermark value", ((ControlTuple.Watermark)controlSink.collectedTuples.get(1)).getTimestamp(), BASE + 800); }
@Test public void testGlobalWindowAssignment() { WindowedOperatorImpl<Long, MutableLong, Long> windowedOperator = createDefaultWindowedOperator(); windowedOperator.setWindowOption(new WindowOption.GlobalWindow()); windowedOperator.setup(testMeta.operatorContext); Tuple.WindowedTuple<Long> windowedValue = windowedOperator.getWindowedValue(new Tuple.TimestampedTuple<>(BASE + 1100L, 2L)); Collection<? extends Window> windows = windowedValue.getWindows(); Assert.assertEquals(1, windows.size()); Assert.assertEquals(Window.GlobalWindow.INSTANCE, windows.iterator().next()); windowedOperator.teardown(); }
@Test public void testTimeWindowAssignment() { WindowedOperatorImpl<Long, MutableLong, Long> windowedOperator = createDefaultWindowedOperator(); windowedOperator.setWindowOption(new WindowOption.TimeWindows(Duration.millis(1000))); windowedOperator.setup(testMeta.operatorContext); Tuple.WindowedTuple<Long> windowedValue = windowedOperator.getWindowedValue(new Tuple.TimestampedTuple<>(BASE + 1100L, 2L)); Collection<? extends Window> windows = windowedValue.getWindows(); Assert.assertEquals(1, windows.size()); Window window = windows.iterator().next(); Assert.assertEquals(BASE + 1000, window.getBeginTimestamp()); Assert.assertEquals(1000, window.getDurationMillis()); }
@Override public MutablePair<MutableLong, MutableLong> accumulate(MutablePair<MutableLong, MutableLong> accumulatedValue, MutablePair<Double, Double> input) { if (input.getLeft() * input.getLeft() + input.getRight() * input.getRight() < 1) { accumulatedValue.getLeft().increment(); } accumulatedValue.getRight().increment(); return accumulatedValue; }
@Override public MutablePair<MutableLong, MutableLong> merge(MutablePair<MutableLong, MutableLong> accumulatedValue1, MutablePair<MutableLong, MutableLong> accumulatedValue2) { accumulatedValue1.getLeft().add(accumulatedValue2.getLeft()); accumulatedValue1.getRight().add(accumulatedValue2.getRight()); return accumulatedValue1; }
@Override public void populateDAG(DAG dag, Configuration configuration) { WordGenerator inputOperator = new WordGenerator(); KeyedWindowedOperatorImpl<String, Long, MutableLong, Long> windowedOperator = new KeyedWindowedOperatorImpl<>(); Accumulation<Long, MutableLong, Long> sum = new SumLong(); windowedOperator.setAccumulation(sum); windowedOperator.setDataStorage(new InMemoryWindowedKeyedStorage<String, MutableLong>()); windowedOperator.setRetractionStorage(new InMemoryWindowedKeyedStorage<String, Long>()); windowedOperator.setWindowStateStorage(new InMemoryWindowedStorage<WindowState>()); windowedOperator.setWindowOption(new WindowOption.TimeWindows(Duration.standardMinutes(1))); windowedOperator.setTriggerOption(TriggerOption.AtWatermark().withEarlyFiringsAtEvery( Duration.millis(1000)).accumulatingAndRetractingFiredPanes()); ConsoleOutputOperator consoleOutput = new ConsoleOutputOperator(); GenericFileOutputOperator<Object> fileOutput = new GenericFileOutputOperator<>(); fileOutput.setConverter(new ToStringConverter()); dag.addOperator("inputOperator", inputOperator); dag.addOperator("windowedOperator", windowedOperator); dag.addOperator("consoleOutput", consoleOutput); dag.addOperator("output", fileOutput); dag.addStream("input_windowed", inputOperator.output, windowedOperator.input); dag.addStream("windowed_output", windowedOperator.output, consoleOutput.input, fileOutput.input); }
