You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2015/03/28 01:05:19 UTC
[42/50] incubator-kylin git commit: refactor
refactor
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/a36d4166
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/a36d4166
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/a36d4166
Branch: refs/heads/streaming-localdict
Commit: a36d416602885ea546ea6a388ba2f3eb78188df7
Parents: ea96dc5
Author: qianhao.zhou <qi...@ebay.com>
Authored: Fri Mar 27 16:55:44 2015 +0800
Committer: qianhao.zhou <qi...@ebay.com>
Committed: Fri Mar 27 16:55:44 2015 +0800
----------------------------------------------------------------------
.../kylin/job/streaming/StreamingBootstrap.java | 5 +-
.../apache/kylin/job/BuildIIWithStreamTest.java | 4 +-
.../job/hadoop/invertedindex/II2CubeTest.java | 58 ++++++++------------
3 files changed, 28 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/a36d4166/job/src/main/java/org/apache/kylin/job/streaming/StreamingBootstrap.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/job/streaming/StreamingBootstrap.java b/job/src/main/java/org/apache/kylin/job/streaming/StreamingBootstrap.java
index 5d1673c..f6abad7 100644
--- a/job/src/main/java/org/apache/kylin/job/streaming/StreamingBootstrap.java
+++ b/job/src/main/java/org/apache/kylin/job/streaming/StreamingBootstrap.java
@@ -123,11 +123,10 @@ public class StreamingBootstrap {
getStreamQueue().put(new Stream(offset, bytes));
}
};
- final IIDesc desc = ii.getDescriptor();
kafkaConsumers.put(getKey(streaming, partitionId), consumer);
- final IIStreamBuilder task = new IIStreamBuilder(consumer.getStreamQueue(), iiSegment.getStorageLocationIdentifier(), desc, partitionId);
- task.setStreamParser(JsonStreamParser.instance);
+ final IIStreamBuilder task = new IIStreamBuilder(consumer.getStreamQueue(), iiSegment.getStorageLocationIdentifier(), iiSegment.getIIInstance(), partitionId);
+ task.setStreamParser(new JsonStreamParser(ii.getDescriptor().listAllColumns()));
Executors.newSingleThreadExecutor().submit(consumer);
Executors.newSingleThreadExecutor().submit(task).get();
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/a36d4166/job/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java b/job/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java
index dae2d03..a3a7489 100644
--- a/job/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java
+++ b/job/src/test/java/org/apache/kylin/job/BuildIIWithStreamTest.java
@@ -215,14 +215,14 @@ public class BuildIIWithStreamTest {
ExecutorService executorService = Executors.newSingleThreadExecutor();
- final IIStreamBuilder streamBuilder = new IIStreamBuilder(queue, segment.getStorageLocationIdentifier(), desc, 0);
+ final IIStreamBuilder streamBuilder = new IIStreamBuilder(queue, segment.getStorageLocationIdentifier(), segment.getIIInstance(), 0);
int count = 0;
while (reader.next()) {
queue.put(parse(reader.getRow()));
count++;
}
logger.info("total record count:" + count + " htable:" + segment.getStorageLocationIdentifier());
- queue.put(new Stream(-1, null));
+ queue.put(Stream.EOF);
final Future<?> future = executorService.submit(streamBuilder);
try {
future.get();
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/a36d4166/job/src/test/java/org/apache/kylin/job/hadoop/invertedindex/II2CubeTest.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/org/apache/kylin/job/hadoop/invertedindex/II2CubeTest.java b/job/src/test/java/org/apache/kylin/job/hadoop/invertedindex/II2CubeTest.java
index 97c71f8..da1cb18 100644
--- a/job/src/test/java/org/apache/kylin/job/hadoop/invertedindex/II2CubeTest.java
+++ b/job/src/test/java/org/apache/kylin/job/hadoop/invertedindex/II2CubeTest.java
@@ -1,14 +1,9 @@
package org.apache.kylin.job.hadoop.invertedindex;
-import java.io.IOException;
-import java.util.*;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import javax.annotation.Nullable;
-
+import com.google.common.base.Function;
+import com.google.common.collect.Collections2;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.MutationSerialization;
@@ -24,22 +19,20 @@ import org.apache.kylin.invertedindex.IIManager;
import org.apache.kylin.invertedindex.index.Slice;
import org.apache.kylin.invertedindex.index.TableRecordInfo;
import org.apache.kylin.invertedindex.index.TableRecordInfoDigest;
-import org.apache.kylin.invertedindex.model.IIDesc;
-import org.apache.kylin.invertedindex.model.IIKeyValueCodecWithState;
-import org.apache.kylin.invertedindex.model.IIRow;
-import org.apache.kylin.invertedindex.model.KeyValueCodec;
+import org.apache.kylin.invertedindex.model.*;
import org.apache.kylin.job.constant.BatchConstants;
import org.apache.kylin.job.hadoop.cube.FactDistinctIIColumnsMapper;
import org.apache.kylin.streaming.Stream;
+import org.apache.kylin.streaming.StringStreamParser;
+import org.apache.kylin.streaming.invertedindex.SliceBuilder;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
-import com.google.common.base.Function;
-import com.google.common.collect.Collections2;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.*;
/**
* Created by Hongbin Ma(Binmahone) on 3/26/15.
@@ -53,10 +46,10 @@ public class II2CubeTest extends LocalFileMetadataTestCase {
List<IIRow> iiRows;
- final String[] inputs = new String[] { //
- "FP-non GTC,0,15,145970,0,28,Toys,2008-10-08 07:18:40,USER_Y,Toys & Hobbies,Models & Kits,Automotive,0,Ebay,USER_S,15,Professional-Other,2012-08-16,2012-08-11,0,2012-08-16,145970,10000329,26.8551,0", //
+ final String[] inputs = new String[]{ //
+ "FP-non GTC,0,15,145970,0,28,Toys,2008-10-08 07:18:40,USER_Y,Toys & Hobbies,Models & Kits,Automotive,0,Ebay,USER_S,15,Professional-Other,2012-08-16,2012-08-11,0,2012-08-16,145970,10000329,26.8551,0", //
"ABIN,0,-99,43479,0,21,Photo,2012-09-11 20:26:04,USER_Y,Cameras & Photo,Film Photography,Other,0,Ebay,USER_S,-99,Not Applicable,2012-08-16,2012-08-11,0,2012-08-16,43479,10000807,26.2474,0", //
- "ABIN,0,16,80053,0,12,Computers,2012-06-19 21:15:09,USER_Y,Computers/Tablets & Networking,MonitorProjectors & Accs,Monitors,0,Ebay,USER_S,16,Consumer-Other,2012-08-16,2012-08-11,0,2012-08-16,80053,10000261,94.2273,0" };
+ "ABIN,0,16,80053,0,12,Computers,2012-06-19 21:15:09,USER_Y,Computers/Tablets & Networking,MonitorProjectors & Accs,Monitors,0,Ebay,USER_S,16,Consumer-Other,2012-08-16,2012-08-11,0,2012-08-16,80053,10000261,94.2273,0"};
@Before
public void setUp() throws Exception {
@@ -64,22 +57,20 @@ public class II2CubeTest extends LocalFileMetadataTestCase {
this.ii = IIManager.getInstance(getTestConfig()).getII(iiName);
this.iiDesc = ii.getDescriptor();
- Collection<?> streams = Collections2.transform(Arrays.asList(inputs), new Function<String, Stream>() {
+ List<Stream> streams = Lists.transform(Arrays.asList(inputs), new Function<String, Stream>() {
@Nullable
@Override
public Stream apply(String input) {
- return new Stream(0, input.getBytes());
+ return new Stream(System.currentTimeMillis(), input.getBytes());
}
});
- LinkedBlockingQueue q = new LinkedBlockingQueue();
- q.addAll(streams);
- q.put(new Stream(-1, null));//a stop sign for builder
iiRows = Lists.newArrayList();
- ToyIIStreamBuilder builder = new ToyIIStreamBuilder(q, iiDesc, 0, iiRows);
- ExecutorService executorService = Executors.newSingleThreadExecutor();
- Future<?> future = executorService.submit(builder);
- future.get();
+ final Slice slice = new SliceBuilder(iiDesc, (short) 0).buildSlice(streams, StringStreamParser.instance);
+ IIKeyValueCodec codec = new IIKeyValueCodec(slice.getInfo());
+ for (IIRow iiRow : codec.encodeKeyValue(slice)) {
+ iiRows.add(iiRow);
+ }
}
@@ -131,15 +122,14 @@ public class II2CubeTest extends LocalFileMetadataTestCase {
@Nullable
@Override
public Pair<ImmutableBytesWritable, Result> apply(@Nullable IIRow input) {
- return new Pair<ImmutableBytesWritable, Result>(new ImmutableBytesWritable(new byte[] { 1 }), Result.create(input.makeCells()));
+ return new Pair<ImmutableBytesWritable, Result>(new ImmutableBytesWritable(new byte[]{1}), Result.create(input.makeCells()));
}
})));
List<Pair<LongWritable, Text>> result = mapDriver.run();
- Set<String> lstgNames = Sets.newHashSet("FP-non GTC","ABIN");
- for(Pair<LongWritable, Text> pair : result)
- {
- Assert.assertEquals(pair.getFirst().get(),6);
+ Set<String> lstgNames = Sets.newHashSet("FP-non GTC", "ABIN");
+ for (Pair<LongWritable, Text> pair : result) {
+ Assert.assertEquals(pair.getFirst().get(), 6);
Assert.assertTrue(lstgNames.contains(pair.getSecond().toString()));
}
}