kylin-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nju_y...@apache.org
Subject [2/2] kylin git commit: APACHE-KYLIN-2732: better way to fix bug of missing records
Date Mon, 04 Dec 2017 16:51:27 GMT
APACHE-KYLIN-2732: better way to fix bug of missing records

Signed-off-by: Zhong <nju_yaho@apache.org>


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

Branch: refs/heads/master
Commit: 5a6e6d7e38342691a057345a9d51c4375df8dc8f
Parents: d103e63
Author: Wang Ken <mingmwang@ebay.com>
Authored: Mon Dec 4 13:13:18 2017 +0800
Committer: Zhong <nju_yaho@apache.org>
Committed: Tue Dec 5 00:38:22 2017 +0800

----------------------------------------------------------------------
 .../ConsumeBlockingQueueController.java         |   2 +-
 .../cube/inmemcubing/DoggedCubeBuilder.java     |  19 ++-
 .../RecordConsumeBlockingQueueController.java   |  43 +++----
 .../ConsumeBlockingQueueControllerTest.java     | 124 +++++++++++++++++++
 .../inmemcubing/ITDoggedCubeBuilderTest.java    |   5 +-
 .../inmemcubing/ITInMemCubeBuilderTest.java     |  13 ++
 6 files changed, 164 insertions(+), 42 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kylin/blob/5a6e6d7e/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/ConsumeBlockingQueueController.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/ConsumeBlockingQueueController.java
b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/ConsumeBlockingQueueController.java
index a9e55f7..81f44a0 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/ConsumeBlockingQueueController.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/ConsumeBlockingQueueController.java
@@ -78,7 +78,7 @@ public class ConsumeBlockingQueueController<T> implements Iterator<T>
{
         hasException = true;
     }
 
-    public long getOutputRowCount() {
+    public int getOutputRowCount() {
         return outputRowCount.get();
     }
 }

http://git-wip-us.apache.org/repos/asf/kylin/blob/5a6e6d7e/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/DoggedCubeBuilder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/DoggedCubeBuilder.java
b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/DoggedCubeBuilder.java
index ccd7137..d761505 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/DoggedCubeBuilder.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/DoggedCubeBuilder.java
@@ -50,7 +50,6 @@ public class DoggedCubeBuilder extends AbstractInMemCubeBuilder {
 
     private static Logger logger = LoggerFactory.getLogger(DoggedCubeBuilder.class);
 
-    private int splitRowThreshold = Integer.MAX_VALUE;
     private int unitRows = ConsumeBlockingQueueController.DEFAULT_BATCH_SIZE;
 
     public DoggedCubeBuilder(CuboidScheduler cuboidScheduler, IJoinedFlatTableDesc flatDesc,
@@ -62,11 +61,6 @@ public class DoggedCubeBuilder extends AbstractInMemCubeBuilder {
             unitRows /= 10;
     }
 
-    public void setSplitRowThreshold(int rowThreshold) {
-        this.splitRowThreshold = rowThreshold;
-        this.unitRows = Math.min(unitRows, rowThreshold);
-    }
-
     @Override
     public <T> void build(BlockingQueue<T> input, InputConverterUnit<T>
inputConverterUnit, ICuboidWriter output)
             throws IOException {
@@ -80,6 +74,9 @@ public class DoggedCubeBuilder extends AbstractInMemCubeBuilder {
 
         public <T> void build(BlockingQueue<T> input, InputConverterUnit<T>
inputConverterUnit, ICuboidWriter output)
                 throws IOException {
+            final RecordConsumeBlockingQueueController<T> inputController = RecordConsumeBlockingQueueController
+                    .getQueueController(inputConverterUnit, input, unitRows);
+
             final List<SplitThread> splits = new ArrayList<SplitThread>();
             final Merger merger = new Merger();
 
@@ -88,8 +85,11 @@ public class DoggedCubeBuilder extends AbstractInMemCubeBuilder {
 
             try {
                 while (true) {
-                    SplitThread last = new SplitThread(splits.size() + 1, RecordConsumeBlockingQueueController
-                            .getQueueController(inputConverterUnit, input, unitRows));
+                    if (inputController.ifEnd()) {
+                        break;
+                    }
+
+                    SplitThread last = new SplitThread(splits.size() + 1, inputController);
                     splits.add(last);
 
                     last.start();
@@ -99,9 +99,6 @@ public class DoggedCubeBuilder extends AbstractInMemCubeBuilder {
                     last.join();
 
                     checkException(splits);
-                    if (last.inputController.ifEnd()) {
-                        break;
-                    }
                 }
 
                 logger.info("Dogged Cube Build splits complete, took " + (System.currentTimeMillis()
- start) + " ms");

http://git-wip-us.apache.org/repos/asf/kylin/blob/5a6e6d7e/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/RecordConsumeBlockingQueueController.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/RecordConsumeBlockingQueueController.java
b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/RecordConsumeBlockingQueueController.java
index 49cbe1f..5fc3e32 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/RecordConsumeBlockingQueueController.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/RecordConsumeBlockingQueueController.java
@@ -31,24 +31,34 @@ public class RecordConsumeBlockingQueueController<T> extends ConsumeBlockingQueu
    
     private T currentObject = null;
     private volatile boolean ifEnd = false;
-    private volatile boolean cut = false;
-    private long outputRowCountCut = 0L;
 
     @Override
-    public boolean hasNext() {
+    public boolean hasNext() { // should be idempotent
+        if (ifEnd) {
+            return false;
+        }
         if (currentObject != null) {
-            return hasNext(currentObject);
+            return true;
         }
         if (!super.hasNext()) {
             return false;
         }
         currentObject = super.next();
-        return hasNext(currentObject);
+
+        if (inputConverterUnit.ifEnd(currentObject)) {
+            ifEnd = true;
+            return false;
+        } else if (inputConverterUnit.ifCut(currentObject)) {
+            currentObject = null;
+            hasNext();
+            return false;
+        }
+        return true;
     }
 
     @Override
     public T next() {
-        if (ifEnd())
+        if (ifEnd() || currentObject == null)
             throw new IllegalStateException();
 
         T result = currentObject;
@@ -59,18 +69,6 @@ public class RecordConsumeBlockingQueueController<T> extends ConsumeBlockingQueu
     public boolean ifEnd() {
         return ifEnd;
     }
-
-    private boolean hasNext(T object) {
-        if (inputConverterUnit.ifEnd(object)) {
-            ifEnd = true;
-            return false;
-        }else if(cut){
-            return false;
-        }else if(inputConverterUnit.ifCut(object)){
-            return false;
-        }
-        return true;
-    }
     
     public static <T> RecordConsumeBlockingQueueController<T> getQueueController(InputConverterUnit<T>
inputConverterUnit, BlockingQueue<T> input){
         return new RecordConsumeBlockingQueueController<>(inputConverterUnit, input,
DEFAULT_BATCH_SIZE);
@@ -79,13 +77,4 @@ public class RecordConsumeBlockingQueueController<T> extends ConsumeBlockingQueu
     public static <T> RecordConsumeBlockingQueueController<T> getQueueController(InputConverterUnit<T>
inputConverterUnit, BlockingQueue<T> input, int batchSize){
         return new RecordConsumeBlockingQueueController<>(inputConverterUnit, input,
batchSize);
     }
-
-    public void forceCutPipe() {
-        cut = true;
-        outputRowCountCut = getOutputRowCount();
-    }
-
-    public long getOutputRowCountAfterCut() {
-        return getOutputRowCount() - outputRowCountCut;
-    }
 }

http://git-wip-us.apache.org/repos/asf/kylin/blob/5a6e6d7e/core-cube/src/test/java/org/apache/kylin/cube/inmemcubing/ConsumeBlockingQueueControllerTest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/test/java/org/apache/kylin/cube/inmemcubing/ConsumeBlockingQueueControllerTest.java
b/core-cube/src/test/java/org/apache/kylin/cube/inmemcubing/ConsumeBlockingQueueControllerTest.java
new file mode 100644
index 0000000..a443596
--- /dev/null
+++ b/core-cube/src/test/java/org/apache/kylin/cube/inmemcubing/ConsumeBlockingQueueControllerTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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.kylin.cube.inmemcubing;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.kylin.gridtable.GTRecord;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ConsumeBlockingQueueControllerTest {
+    private static final Logger logger = LoggerFactory.getLogger(ConsumeBlockingQueueControllerTest.class);
+
+    @Test
+    public void testIterator() {
+        final int nRecord = 4345;
+        final int nCut = 2000;
+        final int nBatch = 60;
+
+        final BlockingQueue<String> input = new LinkedBlockingQueue<>();
+        final RecordConsumeBlockingQueueController inputController = RecordConsumeBlockingQueueController
+                .getQueueController(new InputConverterUnitTest(), input, nBatch);
+
+        new Thread(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    for (int i = 1; i <= nRecord; i++) {
+                        input.put("test");
+                        if (i % nCut == 0) {
+                            input.put(InputConverterUnitTest.CUT_ROW);
+                        }
+                    }
+                    input.put(InputConverterUnitTest.END_ROW);
+                } catch (InterruptedException e) {
+                    logger.warn("Fail to produce records into BlockingQueue due to: " + e);
+                }
+            }
+        }).start();
+
+        final AtomicInteger nRecordConsumed = new AtomicInteger(0);
+        final AtomicInteger nSplit = new AtomicInteger(0);
+        while (true) {
+            // producer done & consume the end row flag
+            if (inputController.ifEnd()) {
+                break;
+            }
+            nSplit.incrementAndGet();
+            Thread consumer = new Thread(new Runnable() {
+                @Override
+                public void run() {
+                    while (inputController.hasNext()) {
+                        inputController.next();
+                        nRecordConsumed.incrementAndGet();
+                    }
+                    System.out.println(nRecordConsumed.get() + " records consumed when finished
split " + nSplit.get());
+                }
+            });
+            consumer.start();
+            try {
+                consumer.join();
+            } catch (InterruptedException e) {
+                logger.warn("Fail to join consumer thread: " + e);
+            }
+        }
+
+        Assert.assertEquals(nRecord, nRecordConsumed.get());
+    }
+
+    private static class InputConverterUnitTest implements InputConverterUnit<String>
{
+        public static final String END_ROW = new String();
+        public static final String CUT_ROW = "0";
+
+        @Override
+        public void convert(String currentObject, GTRecord record) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean ifEnd(String currentObject) {
+            return currentObject == END_ROW;
+        }
+
+        @Override
+        public boolean ifCut(String currentObject) {
+            return currentObject == CUT_ROW;
+        }
+
+        @Override
+        public String getEndRow() {
+            return END_ROW;
+        }
+
+        @Override
+        public String getCutRow() {
+            return CUT_ROW;
+        }
+
+        @Override
+        public boolean ifChange() {
+            throw new UnsupportedOperationException();
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/kylin/blob/5a6e6d7e/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderTest.java
b/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderTest.java
index 0338da8..865cdbb 100644
--- a/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITDoggedCubeBuilderTest.java
@@ -89,12 +89,11 @@ public class ITDoggedCubeBuilderTest extends LocalFileMetadataTestCase
{
         IJoinedFlatTableDesc flatDesc = EngineFactory.getJoinedFlatTableDesc(cube.getDescriptor());
         DoggedCubeBuilder doggedBuilder = new DoggedCubeBuilder(cube.getCuboidScheduler(),
flatDesc, dictionaryMap);
         doggedBuilder.setConcurrentThreads(THREADS);
-        doggedBuilder.setSplitRowThreshold(SPLIT_ROWS);
         FileRecordWriter doggedResult = new FileRecordWriter();
 
         {
             Future<?> future = executorService.submit(doggedBuilder.buildAsRunnable(queue,
doggedResult));
-            ITInMemCubeBuilderTest.feedData(cube, flatTable, queue, INPUT_ROWS, randSeed);
+            ITInMemCubeBuilderTest.feedData(cube, flatTable, queue, INPUT_ROWS, randSeed,
SPLIT_ROWS);
             future.get();
             doggedResult.close();
         }
@@ -110,7 +109,7 @@ public class ITDoggedCubeBuilderTest extends LocalFileMetadataTestCase
{
             inmemResult.close();
         }
 
-        fileCompare(doggedResult.file, inmemResult.file);
+        fileCompare(inmemResult.file, doggedResult.file);
         doggedResult.file.delete();
         inmemResult.file.delete();
     }

http://git-wip-us.apache.org/repos/asf/kylin/blob/5a6e6d7e/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITInMemCubeBuilderTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITInMemCubeBuilderTest.java
b/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITInMemCubeBuilderTest.java
index 3f97f80..f1b65b0 100644
--- a/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITInMemCubeBuilderTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/cube/inmemcubing/ITInMemCubeBuilderTest.java
@@ -151,6 +151,11 @@ public class ITInMemCubeBuilderTest extends LocalFileMetadataTestCase
{
 
     static void feedData(final CubeInstance cube, final String flatTable, ArrayBlockingQueue<String[]>
queue, int count,
             long randSeed) throws IOException, InterruptedException {
+        feedData(cube, flatTable, queue, count, randSeed, Integer.MAX_VALUE);
+    }
+
+    static void feedData(final CubeInstance cube, final String flatTable, ArrayBlockingQueue<String[]>
queue, int count,
+            long randSeed, int splitRowThreshold) throws IOException, InterruptedException
{
         IJoinedFlatTableDesc flatDesc = EngineFactory.getJoinedFlatTableDesc(cube.getDescriptor());
         int nColumns = flatDesc.getAllColumns().size();
 
@@ -178,6 +183,7 @@ public class ITInMemCubeBuilderTest extends LocalFileMetadataTestCase
{
             rand.setSeed(randSeed);
 
         // output with random data
+        int countOfLastSplit = 0;
         for (; count > 0; count--) {
             String[] row = new String[nColumns];
             for (int i = 0; i < nColumns; i++) {
@@ -185,6 +191,13 @@ public class ITInMemCubeBuilderTest extends LocalFileMetadataTestCase
{
                 row[i] = candidates[rand.nextInt(candidates.length)];
             }
             queue.put(row);
+
+            // put cut row if possible
+            countOfLastSplit++;
+            if (countOfLastSplit >= splitRowThreshold) {
+                queue.put(InputConverterUnitForRawData.CUT_ROW);
+                countOfLastSplit = 0;
+            }
         }
         queue.put(InputConverterUnitForRawData.END_ROW);
     }


Mime
View raw message