[
https://issues.apache.org/jira/browse/APEXMALHAR-2100?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15405560#comment-15405560
]
ASF GitHub Bot commented on APEXMALHAR-2100:
--------------------------------------------
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/apex-malhar/pull/330#discussion_r73298635
--- Diff:
library/src/main/java/com/datatorrent/lib/join/AbstractManagedStateInnerJoinOperator.java
---
@@ -0,0 +1,354 @@
+/**
+ * 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 com.datatorrent.lib.join;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+import org.joda.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.apex.malhar.lib.state.managed.ManagedTimeStateImpl;
+import org.apache.apex.malhar.lib.state.managed.ManagedTimeStateMultiValue;
+import org.apache.apex.malhar.lib.state.spillable.Spillable;
+import
org.apache.apex.malhar.lib.state.spillable.SpillableComplexComponent;
+import org.apache.apex.malhar.lib.utils.serde.Serde;
+import org.apache.hadoop.fs.Path;
+import com.google.common.collect.Maps;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.Operator;
+import com.datatorrent.lib.fileaccess.FileAccessFSImpl;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * An abstract implementation of inner join operator over Managed state
which extends from
+ * AbstractInnerJoinOperator.
+ *
+ * <b>Properties:</b><br>
+ * <b>noOfBuckets</b>: Number of buckets required for Managed state. <br>
+ * <b>bucketSpanTime</b>: Indicates the length of the time bucket. <br>
+ */
+@org.apache.hadoop.classification.InterfaceStability.Evolving
+public abstract class AbstractManagedStateInnerJoinOperator<K,T> extends
AbstractInnerJoinOperator<K,T> implements
+ Operator.CheckpointNotificationListener,
Operator.CheckpointListener,Operator.IdleTimeHandler
+{
+ private static final transient Logger LOG =
LoggerFactory.getLogger(AbstractManagedStateInnerJoinOperator.class);
+ public static final String stateDir = "managedState";
+ public static final String stream1State = "stream1Data";
+ public static final String stream2State = "stream2Data";
+ private transient Map<JoinEvent<K,T>, Future<List>> waitingEvents =
Maps.newLinkedHashMap();
+ private int noOfBuckets = 1;
+ private Long bucketSpanTime;
+ protected ManagedTimeStateImpl stream1Store;
+ protected ManagedTimeStateImpl stream2Store;
+
+ /**
+ * Create Managed states and stores for both the streams.
+ */
+ @Override
+ public void createStores()
+ {
+ stream1Store = new ManagedTimeStateImpl();
+ stream2Store = new ManagedTimeStateImpl();
+ stream1Store.setNumBuckets(noOfBuckets);
+ stream2Store.setNumBuckets(noOfBuckets);
+ if (bucketSpanTime != null) {
+
stream1Store.getTimeBucketAssigner().setBucketSpan(Duration.millis(bucketSpanTime));
+
stream2Store.getTimeBucketAssigner().setBucketSpan(Duration.millis(bucketSpanTime));
+ }
+
stream1Store.getTimeBucketAssigner().setExpireBefore(Duration.millis(getExpiryTime()));
+
stream2Store.getTimeBucketAssigner().setExpireBefore(Duration.millis(getExpiryTime()));
+
+ component = new ManagedSpillableComplexComponent();
+ stream1Data =
((ManagedSpillableComplexComponent)component).newSpillableByteArrayListMultimap(stream1Store,
!isStream1KeyPrimary());
+ stream2Data =
((ManagedSpillableComplexComponent)component).newSpillableByteArrayListMultimap(stream2Store,
!isStream2KeyPrimary());
+ }
+
+ /**
+ * Process the tuple which are received from input ports with the
following steps:
+ * 1) Extract key from the given tuple
+ * 2) Insert <key,tuple> into the store where store is the stream1Data
if the tuple
+ * receives from stream1 or viceversa.
+ * 3) Get the values of the key in asynchronous if found it in opposite
store
+ * 4) If the future is done then Merge the given tuple and values found
from step (3) otherwise
+ * put it in waitingEvents
+ * @param tuple given tuple
+ * @param isStream1Data Specifies whether the given tuple belongs to
stream1 or not.
+ */
+ @Override
+ protected void processTuple(T tuple, boolean isStream1Data)
+ {
+ Spillable.SpillableByteArrayListMultimap<K,T> store = isStream1Data ?
stream1Data : stream2Data;
+ K key = extractKey(tuple,isStream1Data);
+ long timeBucket = extractTime(tuple,isStream1Data);
+ if (!((ManagedTimeStateMultiValue)store).put(key, tuple,timeBucket)) {
+ return;
+ }
+ Spillable.SpillableByteArrayListMultimap<K, T> valuestore =
isStream1Data ? stream2Data : stream1Data;
+ Future<List> future =
((ManagedTimeStateMultiValue)valuestore).getAsync(key);
+ if (future.isDone()) {
+ try {
+ joinStream(tuple,isStream1Data, future.get());
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ waitingEvents.put(new JoinEvent<>(key,tuple,isStream1Data),future);
+ }
+ }
+
+ @Override
+ public void handleIdleTime()
+ {
+ if (waitingEvents.size() > 0) {
+ processWaitEvents(false);
+ }
+ }
+
+ @Override
+ public void beforeCheckpoint(long l)
+ {
+ stream1Store.beforeCheckpoint(l);
+ stream2Store.beforeCheckpoint(l);
+ }
+
+ @Override
+ public void checkpointed(long l)
+ {
+ stream1Store.checkpointed(l);
+ stream2Store.checkpointed(l);
+ }
+
+ @Override
+ public void committed(long l)
+ {
+ stream1Store.committed(l);
+ stream2Store.committed(l);
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ super.setup(context);
+
((FileAccessFSImpl)stream1Store.getFileAccess()).setBasePath(context.getValue(DAG.APPLICATION_PATH)
+ Path.SEPARATOR + stateDir + Path.SEPARATOR + String.valueOf(context.getId())
+ Path.SEPARATOR + stream1State);
+
((FileAccessFSImpl)stream2Store.getFileAccess()).setBasePath(context.getValue(DAG.APPLICATION_PATH)
+ Path.SEPARATOR + stateDir + Path.SEPARATOR + String.valueOf(context.getId())
+ Path.SEPARATOR + stream2State);
+ stream1Store.getCheckpointManager().setRecoveryPath("managed_state_" +
stream1State);
+ stream1Store.getCheckpointManager().setRecoveryPath("managed_state_" +
stream2State);
+ stream1Store.setup(context);
+ stream2Store.setup(context);
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ stream1Store.beginWindow(windowId);
+ stream2Store.beginWindow(windowId);
+ super.beginWindow(windowId);
+ }
+
+ /**
+ * Process the waiting events
+ * @param finalize finalize Whether or not to wait for future to return
+ */
+ private void processWaitEvents(boolean finalize)
+ {
+ Iterator<Map.Entry<JoinEvent<K,T>, Future<List>>> waitIterator =
waitingEvents.entrySet().iterator();
+ while (waitIterator.hasNext()) {
+ Map.Entry<JoinEvent<K,T>, Future<List>> waitingEvent =
waitIterator.next();
+ Future<List> future = waitingEvent.getValue();
+ if (future.isDone() || finalize) {
+ try {
+ JoinEvent<K,T> event = waitingEvent.getKey();
+ joinStream(event.value,event.isStream1Data,future.get());
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("end window", e);
+ }
+ waitIterator.remove();
+ if (!finalize) {
+ break;
+ }
+ }
+ }
+ }
+
+ @Override
+ public void endWindow()
+ {
+ processWaitEvents(true);
+ stream1Store.endWindow();
+ stream2Store.endWindow();
+ super.endWindow();
+ }
+
+ @Override
+ public void teardown()
+ {
+ stream1Store.teardown();
+ stream2Store.teardown();
+ super.teardown();
+ }
+
+ /**
+ * Return the number of buckets
+ * @return the noOfBuckets
+ */
+ public int getNoOfBuckets()
+ {
+ return noOfBuckets;
+ }
+
+ /**
+ * Set the number of buckets required for managed state
+ * @param noOfBuckets noOfBuckets
+ */
+ public void setNoOfBuckets(int noOfBuckets)
+ {
+ this.noOfBuckets = noOfBuckets;
+ }
+
+ /**
+ * Return the bucketSpanTime
+ * @return the bucketSpanTime
+ */
+ public Long getBucketSpanTime()
+ {
+ return bucketSpanTime;
+ }
+
+ /**
+ * Sets the length of the time bucket required for managed state.
+ * @param bucketSpanTime given bucketSpanTime
+ */
+ public void setBucketSpanTime(Long bucketSpanTime)
+ {
+ this.bucketSpanTime = bucketSpanTime;
+ }
+
+ public static class JoinEvent<K,T>
+ {
+ public K key;
+ public T value;
+ public boolean isStream1Data;
+
+ public JoinEvent(K key, T value, boolean isStream1Data)
+ {
+ this.key = key;
+ this.value = value;
+ this.isStream1Data = isStream1Data;
+ }
+ }
+
+ public class ManagedSpillableComplexComponent implements
SpillableComplexComponent
--- End diff --
Do you need this complete class to be implemented?
Can you not do:
new ManagedTimeStateMultiValue<K,V>(store, isPrimaryKey);
new ManagedTimeStateMultiValue<K,V>();
Wherever required?
> Development of Inner Join Operator using Spillable Datastructures
> -----------------------------------------------------------------
>
> Key: APEXMALHAR-2100
> URL: https://issues.apache.org/jira/browse/APEXMALHAR-2100
> Project: Apache Apex Malhar
> Issue Type: Task
> Reporter: Chaitanya
> Assignee: Chaitanya
>
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
