You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by qh...@apache.org on 2015/09/25 04:48:09 UTC
[2/4] incubator-kylin git commit: KYLIN-1011 decompose project
streaming
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/LimitedSizeCondition.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/LimitedSizeCondition.java b/streaming/src/main/java/org/apache/kylin/streaming/LimitedSizeCondition.java
deleted file mode 100644
index e5f0fc4..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/LimitedSizeCondition.java
+++ /dev/null
@@ -1,29 +0,0 @@
-package org.apache.kylin.streaming;
-
-/**
- */
-public class LimitedSizeCondition implements BatchCondition {
-
- private final int limit;
- private int count;
-
- public LimitedSizeCondition(int limit) {
- this.limit = limit;
- this.count = 0;
- }
-
- @Override
- public Result apply(ParsedStreamMessage message) {
- if (count < limit) {
- count++;
-
- if (count == limit) {
- return Result.LAST_ACCEPT_FOR_BATCH;
- } else {
- return Result.ACCEPT;
- }
- } else {
- return Result.REJECT;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/MarginCalculator.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/MarginCalculator.java b/streaming/src/main/java/org/apache/kylin/streaming/MarginCalculator.java
deleted file mode 100644
index 855a0f0..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/MarginCalculator.java
+++ /dev/null
@@ -1,83 +0,0 @@
-//package org.apache.kylin.streaming;
-//
-//import java.util.List;
-//import java.util.concurrent.*;
-//
-//import org.apache.kylin.common.util.DateFormat;
-//import org.apache.kylin.common.util.Pair;
-//import org.slf4j.Logger;
-//import org.slf4j.LoggerFactory;
-//
-//import com.google.common.base.Preconditions;
-//import com.google.common.collect.Lists;
-//
-///**
-// */
-//public class MarginCalculator implements Runnable {
-//
-// private static final Logger logger = LoggerFactory.getLogger(MarginCalculator.class);
-//
-//
-// private final String streaming;
-// private final List<BlockingQueue<StreamMessage>> queues;
-// private final List<Pair<Long,Long>> firstAndLastOffsets;
-// private StreamParser streamParser;
-//
-// public MarginCalculator(String streaming, List<BlockingQueue<StreamMessage>> queues, StreamParser streamParser, List<Pair<Long,Long>> firstAndLastOffsets) {
-// Preconditions.checkArgument(queues.size() > 0);
-// this.streaming = streaming;
-// this.queues = queues;
-// this.streamParser = streamParser;
-// this.firstAndLastOffsets = firstAndLastOffsets;
-// }
-//
-// @Override
-// public void run() {
-// try {
-// final int inputCount = queues.size();
-// final ExecutorService executorService = Executors.newFixedThreadPool(inputCount);
-// final CountDownLatch countDownLatch = new CountDownLatch(inputCount);
-// final List<Future<MicroStreamBatch>> futures = Lists.newLinkedList();
-// int partitionId = 0;
-// for (BlockingQueue<StreamMessage> queue : queues) {
-// futures.add(executorService.submit(new StreamFetcher(partitionId, queue, countDownLatch, batchCondition, streamParser)));
-// }
-// countDownLatch.await();
-// List<MicroStreamBatch> batches = Lists.newLinkedList();
-// for (Future<MicroStreamBatch> future : futures) {
-// if (future.get() != null) {
-// batches.add(future.get());
-// } else {
-// logger.warn("EOF encountered, stop streaming");
-// }
-// }
-//
-// MicroStreamBatch batch = batches.get(0);
-// if (batches.size() > 1) {
-// for (int i = 1; i < inputCount; i++) {
-// if (batches.get(i).size() > 0) {
-// batch = MicroStreamBatch.union(batch, batches.get(i));
-// }
-// }
-// }
-// batch.getTimestamp().setFirst(batchCondition.getStartTime());
-// batch.getTimestamp().setSecond(batchCondition.getEndTime());
-//
-// logger.info("Consuming {} messages, covering from {} to {}", new String[] { String.valueOf(batch.size()), DateFormat.formatToTimeStr(batch.getTimestamp().getFirst()), DateFormat.formatToTimeStr(batch.getTimestamp().getSecond()) });
-// long startTime = System.currentTimeMillis();
-// consumer.consume(batch);
-// logger.info("Batch build costs {} milliseconds", System.currentTimeMillis() - startTime);
-// } catch (InterruptedException ie) {
-// throw new RuntimeException("this thread should not be interrupted", ie);
-// } catch (ExecutionException ee) {
-// logger.error("fetch stream error", ee);
-// throw new RuntimeException(ee);
-// } catch (Exception e) {
-// logger.error("error consume batch", e);
-// throw new RuntimeException("error consume batch", e);
-// }
-//
-//
-// }
-//
-//}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatch.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatch.java b/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatch.java
deleted file mode 100644
index 27f817e..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatch.java
+++ /dev/null
@@ -1,99 +0,0 @@
-package org.apache.kylin.streaming;
-
-import java.util.List;
-
-import org.apache.kylin.common.util.Pair;
-
-import com.google.common.collect.Lists;
-
-/**
- */
-public final class MicroStreamBatch {
-
- private final int partitionId;
-
- private final List<List<String>> streams;
-
- private final Pair<Long, Long> timestamp;
-
- private final Pair<Long, Long> offset;
-
- private int rawMessageCount;
-
- public MicroStreamBatch(int partitionId) {
- this.partitionId = partitionId;
- this.streams = Lists.newLinkedList();
- this.timestamp = Pair.newPair(Long.MAX_VALUE, Long.MIN_VALUE);
- this.offset = Pair.newPair(Long.MAX_VALUE, Long.MIN_VALUE);
- }
-
- private MicroStreamBatch(MicroStreamBatch batch) {
- this.partitionId = batch.partitionId;
- this.streams = Lists.newLinkedList(batch.streams);
- this.timestamp = Pair.newPair(batch.timestamp.getFirst(), batch.timestamp.getSecond());
- this.offset = Pair.newPair(batch.offset.getFirst(), batch.offset.getSecond());
- }
-
- public int getPartitionId() {
- return partitionId;
- }
-
- public final List<List<String>> getStreams() {
- return streams;
- }
-
- public final Pair<Long, Long> getTimestamp() {
- return timestamp;
- }
-
- public final Pair<Long, Long> getOffset() {
- return offset;
- }
-
- public final int size() {
- return streams.size();
- }
-
- public final void incRawMessageCount() {
- this.rawMessageCount++;
- }
-
- public final int getRawMessageCount() {
- return this.rawMessageCount;
- }
-
- public final int getFilteredMessageCount() {
- return this.streams.size();
- }
-
- public final void add(ParsedStreamMessage parsedStreamMessage) {
- if (offset.getFirst() > parsedStreamMessage.getOffset()) {
- offset.setFirst(parsedStreamMessage.getOffset());
- }
- if (offset.getSecond() < parsedStreamMessage.getOffset()) {
- offset.setSecond(parsedStreamMessage.getOffset());
- }
- if (timestamp.getFirst() > parsedStreamMessage.getTimestamp()) {
- timestamp.setFirst(parsedStreamMessage.getTimestamp());
- }
- if (timestamp.getSecond() < parsedStreamMessage.getTimestamp()) {
- timestamp.setSecond(parsedStreamMessage.getTimestamp());
- }
- this.streams.add(parsedStreamMessage.getStreamMessage());
- }
-
- public static MicroStreamBatch union(MicroStreamBatch one, MicroStreamBatch another) {
- MicroStreamBatch result = new MicroStreamBatch(one);
- result.streams.addAll(another.streams);
- result.offset.setFirst(Math.min(result.offset.getFirst(), another.offset.getFirst()));
- result.offset.setSecond(Math.max(result.offset.getSecond(), another.offset.getSecond()));
- result.timestamp.setFirst(Math.min(result.timestamp.getFirst(), another.timestamp.getFirst()));
- result.timestamp.setSecond(Math.max(result.timestamp.getSecond(), another.timestamp.getSecond()));
- result.rawMessageCount = one.rawMessageCount + another.rawMessageCount;
- return result;
- }
-
- public boolean isEmpty() {
- return streams.isEmpty();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatchConsumer.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatchConsumer.java b/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatchConsumer.java
deleted file mode 100644
index 37d6076..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/MicroStreamBatchConsumer.java
+++ /dev/null
@@ -1,11 +0,0 @@
-package org.apache.kylin.streaming;
-
-/**
- */
-public interface MicroStreamBatchConsumer {
-
- void consume(MicroStreamBatch microStreamBatch) throws Exception;
-
- void stop();
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/OffsetPeriodCondition.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/OffsetPeriodCondition.java b/streaming/src/main/java/org/apache/kylin/streaming/OffsetPeriodCondition.java
deleted file mode 100644
index 0e4cc3f..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/OffsetPeriodCondition.java
+++ /dev/null
@@ -1,35 +0,0 @@
-package org.apache.kylin.streaming;
-
-/**
- */
-public class OffsetPeriodCondition implements BatchCondition {
-
- private final long startOffset;
- private final long endOffset;
-
- public OffsetPeriodCondition(long startOffset, long endOffset) {
- this.startOffset = startOffset;
- this.endOffset = endOffset;
- }
-
- public long getStartOffset() {
- return startOffset;
- }
-
- public long getEndOffset() {
- return endOffset;
- }
-
- @Override
- public Result apply(ParsedStreamMessage message) {
- final long offset = message.getOffset();
- if (offset < startOffset) {
- return Result.DISCARD;
- } else if (offset < endOffset) {
- return Result.ACCEPT;
- } else {
- return Result.REJECT;
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/OneOffStreamBuilder.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/OneOffStreamBuilder.java b/streaming/src/main/java/org/apache/kylin/streaming/OneOffStreamBuilder.java
deleted file mode 100644
index ae0f70f..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/OneOffStreamBuilder.java
+++ /dev/null
@@ -1,87 +0,0 @@
-package org.apache.kylin.streaming;
-
-import java.util.List;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-
-import org.apache.kylin.common.util.DateFormat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-
-/**
- */
-public class OneOffStreamBuilder implements Runnable {
-
- private static final Logger logger = LoggerFactory.getLogger(OneOffStreamBuilder.class);
-
- private final String streaming;
- private final List<BlockingQueue<StreamMessage>> queues;
- private final MicroStreamBatchConsumer consumer;
- private final TimePeriodCondition batchCondition;
- private StreamParser streamParser;
-
- public OneOffStreamBuilder(String streaming, List<BlockingQueue<StreamMessage>> queues, StreamParser streamParser, MicroStreamBatchConsumer consumer, long startTime, long endTime, long margin) {
- Preconditions.checkArgument(queues.size() > 0);
- this.batchCondition = new TimePeriodCondition(startTime, endTime, margin);
- this.streaming = streaming;
- this.queues = queues;
- this.consumer = Preconditions.checkNotNull(consumer);
- this.streamParser = streamParser;
- }
-
- @Override
- public void run() {
- try {
- final int inputCount = queues.size();
- final ExecutorService executorService = Executors.newFixedThreadPool(inputCount);
- final CountDownLatch countDownLatch = new CountDownLatch(inputCount);
- final List<Future<MicroStreamBatch>> futures = Lists.newLinkedList();
- int partitionId = 0;
- for (BlockingQueue<StreamMessage> queue : queues) {
- futures.add(executorService.submit(new StreamFetcher(partitionId++, queue, countDownLatch, batchCondition, streamParser)));
- }
- countDownLatch.await();
- List<MicroStreamBatch> batches = Lists.newLinkedList();
- for (Future<MicroStreamBatch> future : futures) {
- if (future.get() != null) {
- batches.add(future.get());
- } else {
- logger.warn("EOF encountered, stop streaming");
- }
- }
-
- MicroStreamBatch batch = batches.get(0);
- if (batches.size() > 1) {
- for (int i = 1; i < inputCount; i++) {
- if (batches.get(i).size() > 0) {
- batch = MicroStreamBatch.union(batch, batches.get(i));
- }
- }
- }
- batch.getTimestamp().setFirst(batchCondition.getStartTime());
- batch.getTimestamp().setSecond(batchCondition.getEndTime());
-
- logger.info("Consuming {} messages, covering from {} to {}", new String[] { String.valueOf(batch.size()), DateFormat.formatToTimeStr(batch.getTimestamp().getFirst()), DateFormat.formatToTimeStr(batch.getTimestamp().getSecond()) });
- long startTime = System.currentTimeMillis();
- consumer.consume(batch);
- logger.info("Batch build costs {} milliseconds", System.currentTimeMillis() - startTime);
- } catch (InterruptedException ie) {
- throw new RuntimeException("this thread should not be interrupted", ie);
- } catch (ExecutionException ee) {
- logger.error("fetch stream error", ee);
- throw new RuntimeException(ee);
- } catch (Exception e) {
- logger.error("error consume batch", e);
- throw new RuntimeException("error consume batch", e);
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/ParsedStreamMessage.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/ParsedStreamMessage.java b/streaming/src/main/java/org/apache/kylin/streaming/ParsedStreamMessage.java
deleted file mode 100644
index 87a6ce7..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/ParsedStreamMessage.java
+++ /dev/null
@@ -1,42 +0,0 @@
-package org.apache.kylin.streaming;
-
-import java.util.List;
-
-/**
- */
-public class ParsedStreamMessage {
-
- private final List<String> streamMessage;
-
- private long offset;
-
- private long timestamp;
-
- /**
- * false for pruned messages
- */
- private boolean accepted;
-
- public ParsedStreamMessage(List<String> streamMessage, long offset, long timestamp, boolean accepted) {
- this.streamMessage = streamMessage;
- this.offset = offset;
- this.timestamp = timestamp;
- this.accepted = accepted;
- }
-
- public final List<String> getStreamMessage() {
- return streamMessage;
- }
-
- public final long getOffset() {
- return offset;
- }
-
- public final long getTimestamp() {
- return timestamp;
- }
-
- public final boolean isAccepted() {
- return accepted;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/PartitionMargin.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/PartitionMargin.java b/streaming/src/main/java/org/apache/kylin/streaming/PartitionMargin.java
deleted file mode 100644
index 8f68a0a..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/PartitionMargin.java
+++ /dev/null
@@ -1,13 +0,0 @@
-package org.apache.kylin.streaming;
-
-/**
- */
-public class PartitionMargin {
- public long leftMargin;
- public long rightMargin;
-
- public PartitionMargin(long leftMargin, long rightMargin) {
- this.leftMargin = leftMargin;
- this.rightMargin = rightMargin;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/PartitionMarginCalculator.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/PartitionMarginCalculator.java b/streaming/src/main/java/org/apache/kylin/streaming/PartitionMarginCalculator.java
deleted file mode 100644
index 1951375..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/PartitionMarginCalculator.java
+++ /dev/null
@@ -1,104 +0,0 @@
-package org.apache.kylin.streaming;
-
-import java.util.HashMap;
-import java.util.Queue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-
-/**
- * get the margin for a single partition
- */
-public class PartitionMarginCalculator implements Callable<PartitionMargin> {
-
- /**
- * use the WINDOW_SIZE messages prior to a message to represent its show time
- */
- private class MovingAverage {
-
- private static final int WINDOW_SIZE = 20;
-
- private Queue<Long> q = Lists.newLinkedList();
- private long totalSum = 0;
-
- public long addNewElementAndGetAvg(long e) {
- if (q.size() < WINDOW_SIZE) {
- q.add(e);
- totalSum += e;
- return totalSum / q.size();
- }
-
- long head = q.remove();
- q.add(e);
- return (totalSum - head + e) / WINDOW_SIZE;
- }
- }
-
- private static final Logger logger = LoggerFactory.getLogger(PartitionMarginCalculator.class);
-
- private final BlockingQueue<StreamMessage> streamMessageQueue;
- private final CountDownLatch countDownLatch;
- private final int partitionId;
- private final long lastOffset;
- private final StreamParser streamParser;
-
- //runtime calculations
- private final HashMap<Long, Long> earliestOffsets = Maps.newHashMap();//for each second in context, record its earliest show offset in kafka
- private long maxMargin = 0;
- private final MovingAverage average = new MovingAverage();
-
- public PartitionMarginCalculator(int partitionId, BlockingQueue<StreamMessage> streamMessageQueue, CountDownLatch countDownLatch, long lastOffset, StreamParser streamParser) {
- this.partitionId = partitionId;
- this.streamMessageQueue = streamMessageQueue;
- this.countDownLatch = countDownLatch;
- this.streamParser = streamParser;
- this.lastOffset = lastOffset;
- }
-
- @Override
- public PartitionMargin call() throws Exception {
- try {
- while (true) {
- StreamMessage streamMessage = streamMessageQueue.poll(30, TimeUnit.SECONDS);
- if (streamMessage == null) {
- logger.info("The stream queue for partition {} is drained", partitionId);
- continue;
- }
-
- final ParsedStreamMessage parsedStreamMessage = streamParser.parse(streamMessage);
- if (parsedStreamMessage == null) {
- throw new RuntimeException("parsedStreamMessage of " + new String(streamMessage.getRawData()) + " is null");
- }
-
- if (parsedStreamMessage.getOffset() >= lastOffset) {
- logger.info("The final max margin for partition {} is {} ", partitionId, maxMargin);
- return new PartitionMargin(maxMargin, maxMargin);
- }
-
- long timestamp = parsedStreamMessage.getTimestamp();
- long wallTime = average.addNewElementAndGetAvg(timestamp);
- long formalizedTs = timestamp / 1000 * 1000;
-
- if (earliestOffsets.containsKey(formalizedTs)) {
- this.maxMargin = Math.max(this.maxMargin, Math.abs(earliestOffsets.get(formalizedTs) - wallTime));
- } else {
- earliestOffsets.put(formalizedTs, wallTime);
- }
-
- }
- } catch (Exception e) {
- logger.error("partition margin calculation stream error, stopping", e);
- throw new RuntimeException("partition margin calculation stream error, stopping", e);
- } finally {
- logger.info("one partition sign off");
- countDownLatch.countDown();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/StreamBuilder.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/StreamBuilder.java b/streaming/src/main/java/org/apache/kylin/streaming/StreamBuilder.java
deleted file mode 100644
index 82b6c02..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/StreamBuilder.java
+++ /dev/null
@@ -1,202 +0,0 @@
-/*
- *
- *
- * 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.streaming;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.util.DateFormat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-
-/**
- */
-public class StreamBuilder implements Runnable {
-
- private static final Logger logger = LoggerFactory.getLogger(StreamBuilder.class);
-
- private StreamParser streamParser = StringStreamParser.instance;
-
- private final List<BlockingQueue<StreamMessage>> streamMessageQueues;
-
- private final MicroStreamBatchConsumer consumer;
-
- private final String streaming;
-
- private final long startTimestamp;
-
- private final long batchInterval;
-
- private final int batchSize;
-
- public static final StreamBuilder newPeriodicalStreamBuilder(String streaming, List<BlockingQueue<StreamMessage>> inputs, MicroStreamBatchConsumer consumer, long startTimestamp, long batchInterval) {
- return new StreamBuilder(streaming, inputs, consumer, startTimestamp, batchInterval);
- }
-
- public static final StreamBuilder newLimitedSizeStreamBuilder(String streaming, BlockingQueue<StreamMessage> input, MicroStreamBatchConsumer consumer, long startTimestamp, int batchSize) {
- return new StreamBuilder(streaming, input, consumer, startTimestamp, batchSize);
- }
-
- private StreamBuilder(String streaming, List<BlockingQueue<StreamMessage>> inputs, MicroStreamBatchConsumer consumer, long startTimestamp, long batchInterval) {
- Preconditions.checkArgument(inputs.size() > 0);
- this.streaming = streaming;
- this.streamMessageQueues = Lists.newArrayList();
- this.consumer = Preconditions.checkNotNull(consumer);
- this.startTimestamp = startTimestamp;
- this.batchInterval = batchInterval;
- this.batchSize = -1;
- init(inputs);
- }
-
- private StreamBuilder(String streaming, BlockingQueue<StreamMessage> input, MicroStreamBatchConsumer consumer, long startTimestamp, int batchSize) {
- this.streaming = streaming;
- this.streamMessageQueues = Lists.newArrayList();
- this.consumer = Preconditions.checkNotNull(consumer);
- this.startTimestamp = startTimestamp;
- this.batchInterval = -1L;
- this.batchSize = batchSize;
- init(Preconditions.checkNotNull(input));
- }
-
- private void init(BlockingQueue<StreamMessage> input) {
- this.streamMessageQueues.add(input);
- }
-
- private void init(List<BlockingQueue<StreamMessage>> inputs) {
- this.streamMessageQueues.addAll(inputs);
- }
-
- private BatchCondition generateBatchCondition(long startTimestamp) {
- if (batchInterval > 0) {
- return new TimePeriodCondition(startTimestamp, startTimestamp + batchInterval);
- } else {
- return new LimitedSizeCondition(batchSize);
- }
- }
-
- @Override
- public void run() {
- try {
- final int inputCount = streamMessageQueues.size();
- final ExecutorService executorService = Executors.newFixedThreadPool(inputCount);
- long start = startTimestamp;
- List<Integer> partitions = Lists.newArrayList();
- for (int i = 0, partitionCount = streamMessageQueues.size(); i < partitionCount; i++) {
- partitions.add(i);
- }
- while (true) {
- CountDownLatch countDownLatch = new CountDownLatch(inputCount);
- ArrayList<Future<MicroStreamBatch>> futures = Lists.newArrayListWithExpectedSize(inputCount);
- int partitionId = 0;
- for (BlockingQueue<StreamMessage> streamMessageQueue : streamMessageQueues) {
- futures.add(executorService.submit(new StreamFetcher(partitionId++, streamMessageQueue, countDownLatch, generateBatchCondition(start), getStreamParser())));
- }
- countDownLatch.await();
- ArrayList<MicroStreamBatch> batches = Lists.newArrayListWithExpectedSize(inputCount);
- for (Future<MicroStreamBatch> future : futures) {
- if (future.get() != null) {
- batches.add(future.get());
- } else {
- //EOF occurs, stop consumer
- consumer.stop();
- return;
- }
- }
- MicroStreamBatch batch = batches.get(0);
- if (batches.size() > 1) {
- for (int i = 1; i < inputCount; i++) {
- if (batches.get(i).size() > 0) {
- batch = MicroStreamBatch.union(batch, batches.get(i));
- }
- }
- }
- if (batches.size() > 1) {
- batch.getTimestamp().setFirst(start);
- batch.getTimestamp().setSecond(start + batchInterval);
- }
- if (batchInterval > 0) {
- start += batchInterval;
- }
-
- if (batch.size() == 0) {
- logger.info("nothing to build, skip to next iteration after sleeping 10s");
- Thread.sleep(10000);
- continue;
- } else {
- logger.info("Consuming {} messages, covering from {} to {}", new String[] { String.valueOf(batch.size()), DateFormat.formatToTimeStr(batch.getTimestamp().getFirst()), DateFormat.formatToTimeStr(batch.getTimestamp().getSecond()) });
- long startTime = System.currentTimeMillis();
- consumer.consume(batch);
- logger.info("Batch build costs {} milliseconds", System.currentTimeMillis() - startTime);
- if (batches.size() > 1) {
- final HashMap<Integer, Long> offset = Maps.newHashMap();
- for (MicroStreamBatch microStreamBatch : batches) {
- offset.put(microStreamBatch.getPartitionId(), microStreamBatch.getOffset().getSecond());
- }
- StreamingManager.getInstance(KylinConfig.getInstanceFromEnv()).updateOffset(streaming, offset);
- }
- }
- }
- } catch (InterruptedException e) {
- throw new RuntimeException("stream fetcher thread should not be interrupted", e);
- } catch (ExecutionException e) {
- logger.error("stream fetch thread encountered exception", e);
- throw new RuntimeException("stream fetch thread encountered exception", e);
- } catch (Exception e) {
- logger.error("consumer encountered exception", e);
- throw new RuntimeException("consumer encountered exception", e);
- }
- }
-
- public final StreamParser getStreamParser() {
- return streamParser;
- }
-
- public final void setStreamParser(StreamParser streamParser) {
- this.streamParser = streamParser;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/StreamFetcher.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/StreamFetcher.java b/streaming/src/main/java/org/apache/kylin/streaming/StreamFetcher.java
deleted file mode 100644
index f429a49..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/StreamFetcher.java
+++ /dev/null
@@ -1,109 +0,0 @@
-package org.apache.kylin.streaming;
-
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-
-/**
- */
-public class StreamFetcher implements Callable<MicroStreamBatch> {
- private static final Logger logger = LoggerFactory.getLogger(StreamFetcher.class);
-
- private final BlockingQueue<StreamMessage> streamMessageQueue;
- private final CountDownLatch countDownLatch;
- private final int partitionId;
- private final BatchCondition condition;
- private final StreamParser streamParser;
-
- public StreamFetcher(int partitionId, BlockingQueue<StreamMessage> streamMessageQueue, CountDownLatch countDownLatch, BatchCondition condition, StreamParser streamParser) {
- this.partitionId = partitionId;
- this.streamMessageQueue = streamMessageQueue;
- this.countDownLatch = countDownLatch;
- this.condition = condition;
- this.streamParser = streamParser;
- }
-
- private StreamMessage peek(BlockingQueue<StreamMessage> queue, long timeout) {
- long t = System.currentTimeMillis();
- while (true) {
- final StreamMessage peek = queue.peek();
- if (peek != null) {
- return peek;
- } else {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- logger.warn("stream queue should not be interrupted", e);
- return null;
- }
- if (System.currentTimeMillis() - t > timeout) {
- break;
- }
- }
- }
- return queue.peek();
- }
-
- @Override
- public MicroStreamBatch call() throws Exception {
- try {
- MicroStreamBatch microStreamBatch = null;
- while (true) {
- if (microStreamBatch == null) {
- microStreamBatch = new MicroStreamBatch(partitionId);
- }
- StreamMessage streamMessage = peek(streamMessageQueue, 60000);
- if (streamMessage == null) {
- logger.info("The stream queue is drained, current available stream count: " + microStreamBatch.size());
- return microStreamBatch;
- }
- if (streamMessage.getOffset() < 0) {
- logger.warn("streaming encountered EOF, stop building");
- return null;
- }
-
- microStreamBatch.incRawMessageCount();
- final ParsedStreamMessage parsedStreamMessage = streamParser.parse(streamMessage);
- if (parsedStreamMessage == null) {
- throw new RuntimeException("parsedStreamMessage of " + new String(streamMessage.getRawData()) + " is null");
- }
-
- final BatchCondition.Result result = condition.apply(parsedStreamMessage);
- if (parsedStreamMessage.isAccepted()) {
- if (result == BatchCondition.Result.ACCEPT) {
- streamMessageQueue.take();
- microStreamBatch.add(parsedStreamMessage);
- } else if (result == BatchCondition.Result.LAST_ACCEPT_FOR_BATCH) {
- streamMessageQueue.take();
- microStreamBatch.add(parsedStreamMessage);
- break;
- } else if (result == BatchCondition.Result.DISCARD) {
- streamMessageQueue.take();
- } else if (result == BatchCondition.Result.REJECT) {
- logger.info("Partition :" + partitionId + " rejecting message at " + parsedStreamMessage.getOffset());
- break;
- }
- } else {
- streamMessageQueue.take();
- }
- }
-
- Preconditions.checkArgument(microStreamBatch != null, "microStreamBatch is null!");
- logger.info(String.format("Partition %d contributing %d filtered messages out from %d raw messages"//
- , partitionId, microStreamBatch.getFilteredMessageCount(), microStreamBatch.getRawMessageCount()));
- return microStreamBatch;
-
- } catch (Exception e) {
- logger.error("build stream error, stop building", e);
- throw new RuntimeException("build stream error, stop building", e);
- } finally {
- logger.info("partition {} sign off", partitionId);
- countDownLatch.countDown();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/StreamMessage.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/StreamMessage.java b/streaming/src/main/java/org/apache/kylin/streaming/StreamMessage.java
deleted file mode 100644
index 6229aa5..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/StreamMessage.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- *
- *
- * 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.streaming;
-
-/**
- */
-public class StreamMessage {
-
- private long offset;
- private byte[] rawData;
-
- public static final StreamMessage EOF = new StreamMessage(-1, new byte[0]);
-
- public StreamMessage(long offset, byte[] rawData) {
- this.offset = offset;
- this.rawData = rawData;
- }
-
- public byte[] getRawData() {
- return rawData;
- }
-
- public long getOffset() {
- return offset;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/StreamParser.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/StreamParser.java b/streaming/src/main/java/org/apache/kylin/streaming/StreamParser.java
deleted file mode 100644
index 6a2227d..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/StreamParser.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- *
- *
- * 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.streaming;
-
-/**
- */
-public interface StreamParser {
-
- /**
- * @param streamMessage
- * @return ParsedStreamMessage must not be NULL
- */
- ParsedStreamMessage parse(StreamMessage streamMessage);
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/StreamingConfig.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/StreamingConfig.java b/streaming/src/main/java/org/apache/kylin/streaming/StreamingConfig.java
deleted file mode 100644
index c2d5361..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/StreamingConfig.java
+++ /dev/null
@@ -1,189 +0,0 @@
-/*
- *
- *
- * 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.streaming;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.util.List;
-
-import org.apache.kylin.common.persistence.JsonSerializer;
-import org.apache.kylin.common.persistence.RootPersistentEntity;
-import org.apache.kylin.common.persistence.Serializer;
-
-import com.fasterxml.jackson.annotation.JsonAutoDetect;
-import com.fasterxml.jackson.annotation.JsonManagedReference;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-/**
- */
-@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.NONE, getterVisibility = JsonAutoDetect.Visibility.NONE, isGetterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE)
-public class StreamingConfig extends RootPersistentEntity {
-
- public static Serializer<StreamingConfig> SERIALIZER = new JsonSerializer<StreamingConfig>(StreamingConfig.class);
-
- @JsonProperty("name")
- private String name;
-
- @JsonManagedReference
- @JsonProperty("clusters")
- private List<KafkaClusterConfig> kafkaClusterConfigs;
-
- @JsonProperty("topic")
- private String topic;
-
- @JsonProperty("timeout")
- private int timeout;
-
- @JsonProperty("maxReadCount")
- private int maxReadCount;
-
- @JsonProperty("bufferSize")
- private int bufferSize;
-
- @JsonProperty("iiName")
- private String iiName;
-
- @JsonProperty("cubeName")
- private String cubeName;
-
- @JsonProperty("parserName")
- private String parserName;
-
- //"tsColName=timestamp;x=y"
- @JsonProperty("parserProperties")
- private String parserProperties;
-
- @JsonProperty("margin")
- private long margin;
-
- public List<KafkaClusterConfig> getKafkaClusterConfigs() {
- return kafkaClusterConfigs;
- }
-
- public String getParserName() {
- return parserName;
- }
-
- public void setParserName(String parserName) {
- this.parserName = parserName;
- }
-
- public int getTimeout() {
- return timeout;
- }
-
- public void setTimeout(int timeout) {
- this.timeout = timeout;
- }
-
- public int getMaxReadCount() {
- return maxReadCount;
- }
-
- public void setMaxReadCount(int maxReadCount) {
- this.maxReadCount = maxReadCount;
- }
-
- public int getBufferSize() {
- return bufferSize;
- }
-
- public void setBufferSize(int bufferSize) {
- this.bufferSize = bufferSize;
- }
-
- public String getTopic() {
- return topic;
- }
-
- public void setTopic(String topic) {
- this.topic = topic;
- }
-
- public String getCubeName() {
- return cubeName;
- }
-
- public void setCubeName(String cubeName) {
- this.cubeName = cubeName;
- }
-
- public String getIiName() {
- return iiName;
- }
-
- public void setIiName(String iiName) {
- this.iiName = iiName;
- }
-
- public String getName() {
- return name;
- }
-
- public void setName(String name) {
- this.name = name;
- }
-
- public long getMargin() {
- return margin;
- }
-
- public void setMargin(long margin) {
- this.margin = margin;
- }
-
- public String getParserProperties() {
- return parserProperties;
- }
-
- public void setParserProperties(String parserProperties) {
- this.parserProperties = parserProperties;
- }
-
- @Override
- public StreamingConfig clone() {
- try {
- final ByteArrayOutputStream baos = new ByteArrayOutputStream();
- SERIALIZER.serialize(this, new DataOutputStream(baos));
- return SERIALIZER.deserialize(new DataInputStream(new ByteArrayInputStream(baos.toByteArray())));
- } catch (IOException e) {
- throw new RuntimeException(e);//in mem, should not happen
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/StreamingManager.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/StreamingManager.java b/streaming/src/main/java/org/apache/kylin/streaming/StreamingManager.java
deleted file mode 100644
index 53e5b06..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/StreamingManager.java
+++ /dev/null
@@ -1,221 +0,0 @@
-/*
- *
- *
- * 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.streaming;
-
-import java.io.BufferedReader;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.persistence.ResourceStore;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.fasterxml.jackson.databind.JavaType;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.type.MapType;
-import com.fasterxml.jackson.databind.type.SimpleType;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-
-/**
- */
-public class StreamingManager {
-
- private static final Logger logger = LoggerFactory.getLogger(StreamingManager.class);
-
- // static cached instances
- private static final ConcurrentHashMap<KylinConfig, StreamingManager> CACHE = new ConcurrentHashMap<KylinConfig, StreamingManager>();
-
- private KylinConfig config;
-
- private StreamingManager(KylinConfig config) {
- this.config = config;
- }
-
- private ResourceStore getStore() {
- return ResourceStore.getStore(this.config);
- }
-
- public static StreamingManager getInstance(KylinConfig config) {
- StreamingManager r = CACHE.get(config);
- if (r != null) {
- return r;
- }
-
- synchronized (StreamingManager.class) {
- r = CACHE.get(config);
- if (r != null) {
- return r;
- }
- r = new StreamingManager(config);
- CACHE.put(config, r);
- if (CACHE.size() > 1) {
- logger.warn("More than one cubemanager singleton exist");
- }
- return r;
- }
- }
-
- private boolean checkExistence(String name) {
- return true;
- }
-
- private String formatStreamingConfigPath(String name) {
- return ResourceStore.STREAMING_RESOURCE_ROOT + "/" + name + ".json";
- }
-
- private String formatStreamingOutputPath(String streaming, int partition) {
- return ResourceStore.STREAMING_OUTPUT_RESOURCE_ROOT + "/" + streaming + "_" + partition + ".json";
- }
-
- private String formatStreamingOutputPath(String streaming, List<Integer> partitions) {
- return ResourceStore.STREAMING_OUTPUT_RESOURCE_ROOT + "/" + streaming + "_" + StringUtils.join(partitions, "_") + ".json";
- }
-
- public boolean createOrUpdateKafkaConfig(String name, StreamingConfig config) {
- try {
- getStore().putResource(formatStreamingConfigPath(name), config, StreamingConfig.SERIALIZER);
- return true;
- } catch (IOException e) {
- logger.error("error save resource name:" + name, e);
- throw new RuntimeException("error save resource name:" + name, e);
- }
- }
-
- public StreamingConfig getStreamingConfig(String name) {
- try {
- return getStore().getResource(formatStreamingConfigPath(name), StreamingConfig.class, StreamingConfig.SERIALIZER);
- } catch (IOException e) {
- logger.error("error get resource name:" + name, e);
- throw new RuntimeException("error get resource name:" + name, e);
- }
- }
-
- public void saveStreamingConfig(StreamingConfig streamingConfig) throws IOException {
- if (streamingConfig == null || StringUtils.isEmpty(streamingConfig.getName())) {
- throw new IllegalArgumentException();
- }
-
- String path = formatStreamingConfigPath(streamingConfig.getName());
- getStore().putResource(path, streamingConfig, StreamingConfig.SERIALIZER);
- }
-
- public long getOffset(String streaming, int shard) {
- final String resPath = formatStreamingOutputPath(streaming, shard);
- try {
- final InputStream inputStream = getStore().getResource(resPath);
- if (inputStream == null) {
- return 0;
- } else {
- final BufferedReader br = new BufferedReader(new InputStreamReader(inputStream));
- long ret = Long.parseLong(br.readLine());
- br.close();
- return ret;
- }
- } catch (Exception e) {
- logger.error("error get offset, path:" + resPath, e);
- throw new RuntimeException("error get offset, path:" + resPath, e);
- }
- }
-
- public void updateOffset(String streaming, int shard, long offset) {
- Preconditions.checkArgument(offset >= 0, "offset cannot be smaller than 0");
- final String resPath = formatStreamingOutputPath(streaming, shard);
- try {
- ByteArrayInputStream input = new ByteArrayInputStream(Long.valueOf(offset).toString().getBytes());
- getStore().putResource(resPath, input, getStore().getResourceTimestamp(resPath));
- input.close();
- } catch (IOException e) {
- logger.error("error update offset, path:" + resPath, e);
- throw new RuntimeException("error update offset, path:" + resPath, e);
- }
- }
-
- public Map<Integer, Long> getOffset(String streaming, List<Integer> partitions) {
- Collections.sort(partitions);
- final String resPath = formatStreamingOutputPath(streaming, partitions);
- try {
- final InputStream inputStream = getStore().getResource(resPath);
- if (inputStream == null) {
- return Collections.emptyMap();
- }
-
- HashMap<Integer, Long> result = null;
- try {
- result = mapper.readValue(inputStream, mapType);
- } finally {
- inputStream.close();
- }
-
- return result;
- } catch (IOException e) {
- logger.error("error get offset, path:" + resPath, e);
- throw new RuntimeException("error get offset, path:" + resPath, e);
- }
- }
-
- public void updateOffset(String streaming, HashMap<Integer, Long> offset) {
- List<Integer> partitions = Lists.newLinkedList(offset.keySet());
- Collections.sort(partitions);
- final String resPath = formatStreamingOutputPath(streaming, partitions);
- try {
- final ByteArrayOutputStream baos = new ByteArrayOutputStream();
- mapper.writeValue(baos, offset);
- baos.close();
-
- ByteArrayInputStream input = new ByteArrayInputStream(baos.toByteArray());
- getStore().putResource(resPath, input, getStore().getResourceTimestamp(resPath));
- input.close();
-
- } catch (IOException e) {
- logger.error("error update offset, path:" + resPath, e);
- throw new RuntimeException("error update offset, path:" + resPath, e);
- }
- }
-
- private final ObjectMapper mapper = new ObjectMapper();
- private final JavaType mapType = MapType.construct(HashMap.class, SimpleType.construct(Integer.class), SimpleType.construct(Long.class));
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/StreamingUtil.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/StreamingUtil.java b/streaming/src/main/java/org/apache/kylin/streaming/StreamingUtil.java
deleted file mode 100644
index 1d6b95c..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/StreamingUtil.java
+++ /dev/null
@@ -1,164 +0,0 @@
-package org.apache.kylin.streaming;
-
-import java.nio.ByteBuffer;
-import java.util.Iterator;
-import java.util.Map;
-
-import kafka.api.OffsetRequest;
-import kafka.cluster.Broker;
-import kafka.javaapi.FetchResponse;
-import kafka.javaapi.PartitionMetadata;
-import kafka.message.MessageAndOffset;
-
-import org.apache.kylin.common.util.Pair;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Maps;
-
-/**
- */
-public final class StreamingUtil {
-
- private static final Logger logger = LoggerFactory.getLogger(StreamingUtil.class);
-
- private static final int MAX_RETRY_TIMES = 6;
-
- private StreamingUtil() {
- }
-
- public static Broker getLeadBroker(KafkaClusterConfig kafkaClusterConfig, int partitionId) {
- final PartitionMetadata partitionMetadata = KafkaRequester.getPartitionMetadata(kafkaClusterConfig.getTopic(), partitionId, kafkaClusterConfig.getBrokers(), kafkaClusterConfig);
- if (partitionMetadata != null && partitionMetadata.errorCode() == 0) {
- return partitionMetadata.leader();
- } else {
- return null;
- }
- }
-
- private static void sleep(int retryTimes) {
- int seconds = (int) Math.pow(2, retryTimes);
- logger.info("retry times:" + retryTimes + " sleep:" + seconds + " seconds");
- try {
- Thread.sleep(seconds * 1000);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
- }
-
- private static MessageAndOffset getKafkaMessage(KafkaClusterConfig kafkaClusterConfig, int partitionId, long offset) {
- final String topic = kafkaClusterConfig.getTopic();
- int retry = 0;
- while (retry < MAX_RETRY_TIMES) {//max sleep time 63 seconds
- final Broker leadBroker = getLeadBroker(kafkaClusterConfig, partitionId);
- if (leadBroker == null) {
- logger.warn("unable to find leadBroker with config:" + kafkaClusterConfig + " partitionId:" + partitionId);
- sleep(retry++);
- continue;
- }
- final FetchResponse response = KafkaRequester.fetchResponse(topic, partitionId, offset, leadBroker, kafkaClusterConfig);
- if (response.errorCode(topic, partitionId) != 0) {
- logger.warn("errorCode of FetchResponse is:" + response.errorCode(topic, partitionId));
- sleep(retry++);
- continue;
- }
- final Iterator<MessageAndOffset> iterator = response.messageSet(topic, partitionId).iterator();
- if (!iterator.hasNext()) {
- logger.warn("messageSet is empty");
- sleep(retry++);
- continue;
- }
- return iterator.next();
- }
-
- //going to fail, check start&end offset again:
- try {
- Pair<Long, Long> newStartEnd = getFirstAndLastOffset(kafkaClusterConfig, partitionId);
- logger.info("Partition: {}, latest start offset: {} ,latest end offset: {}", new Object[] { partitionId, newStartEnd.getFirst(), newStartEnd.getSecond() });
- } catch (Exception e) {
- logger.warn("something went wrong", e);
- //just swallow the exception
- }
-
- throw new IllegalStateException(String.format("try to get timestamp of topic: %s, partitionId: %d, offset: %d, failed to get StreamMessage from kafka", topic, partitionId, offset));
- }
-
- public static long findClosestOffsetWithDataTimestamp(KafkaClusterConfig kafkaClusterConfig, int partitionId, long timestamp, StreamParser streamParser) {
- Pair<Long, Long> firstAndLast = getFirstAndLastOffset(kafkaClusterConfig, partitionId);
- final String topic = kafkaClusterConfig.getTopic();
-
- logger.info(String.format("topic: %s, partitionId: %d, try to find closest offset with timestamp: %d between offset {%d, %d}", topic, partitionId, timestamp, firstAndLast.getFirst(), firstAndLast.getSecond()));
- final long result = binarySearch(kafkaClusterConfig, partitionId, firstAndLast.getFirst(), firstAndLast.getSecond(), timestamp, streamParser);
- logger.info(String.format("topic: %s, partitionId: %d, found offset: %d", topic, partitionId, result));
- return result;
- }
-
- public static Pair<Long, Long> getFirstAndLastOffset(KafkaClusterConfig kafkaClusterConfig, int partitionId) {
- final String topic = kafkaClusterConfig.getTopic();
- final Broker leadBroker = Preconditions.checkNotNull(getLeadBroker(kafkaClusterConfig, partitionId), "unable to find leadBroker with config:" + kafkaClusterConfig + " partitionId:" + partitionId);
- final long earliestOffset = KafkaRequester.getLastOffset(topic, partitionId, OffsetRequest.EarliestTime(), leadBroker, kafkaClusterConfig);
- final long latestOffset = KafkaRequester.getLastOffset(topic, partitionId, OffsetRequest.LatestTime(), leadBroker, kafkaClusterConfig) - 1;
- return Pair.newPair(earliestOffset, latestOffset);
- }
-
- private static long binarySearch(KafkaClusterConfig kafkaClusterConfig, int partitionId, long startOffset, long endOffset, long targetTimestamp, StreamParser streamParser) {
- Map<Long, Long> cache = Maps.newHashMap();
-
- while (startOffset < endOffset) {
- long midOffset = startOffset + ((endOffset - startOffset) >> 1);
- long startTimestamp = getDataTimestamp(kafkaClusterConfig, partitionId, startOffset, streamParser, cache);
- long endTimestamp = getDataTimestamp(kafkaClusterConfig, partitionId, endOffset, streamParser, cache);
- long midTimestamp = getDataTimestamp(kafkaClusterConfig, partitionId, midOffset, streamParser, cache);
- // hard to ensure these 2 conditions
- // Preconditions.checkArgument(startTimestamp <= midTimestamp);
- // Preconditions.checkArgument(midTimestamp <= endTimestamp);
- if (startTimestamp >= targetTimestamp) {
- return startOffset;
- }
- if (endTimestamp <= targetTimestamp) {
- return endOffset;
- }
- if (targetTimestamp == midTimestamp) {
- return midOffset;
- } else if (targetTimestamp < midTimestamp) {
- endOffset = midOffset - 1;
- continue;
- } else {
- startOffset = midOffset + 1;
- continue;
- }
- }
- return startOffset;
- }
-
- private static long getDataTimestamp(KafkaClusterConfig kafkaClusterConfig, int partitionId, long offset, StreamParser streamParser, Map<Long, Long> cache) {
- if (cache.containsKey(offset)) {
- return cache.get(offset);
- } else {
- long t = getDataTimestamp(kafkaClusterConfig, partitionId, offset, streamParser);
- cache.put(offset, t);
- return t;
- }
- }
-
- public static long getDataTimestamp(KafkaClusterConfig kafkaClusterConfig, int partitionId, long offset, StreamParser streamParser) {
- final String topic = kafkaClusterConfig.getTopic();
- final MessageAndOffset messageAndOffset = getKafkaMessage(kafkaClusterConfig, partitionId, offset);
- final ByteBuffer payload = messageAndOffset.message().payload();
- byte[] bytes = new byte[payload.limit()];
- payload.get(bytes);
- final ParsedStreamMessage parsedStreamMessage = streamParser.parse(new StreamMessage(messageAndOffset.offset(), bytes));
- logger.debug(String.format("The timestamp of topic: %s, partitionId: %d, offset: %d is: %d", topic, partitionId, offset, parsedStreamMessage.getTimestamp()));
- return parsedStreamMessage.getTimestamp();
-
- }
-
- public static void main(String[] args) {
- if (args == null || args.length == 0) {
- }
-
- if ("calculatemargin".equals(args[0])) {
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/StringStreamParser.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/StringStreamParser.java b/streaming/src/main/java/org/apache/kylin/streaming/StringStreamParser.java
deleted file mode 100644
index 268725f..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/StringStreamParser.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- *
- *
- * 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.streaming;
-
-import com.google.common.collect.Lists;
-
-/**
- */
-public final class StringStreamParser implements StreamParser {
-
- public static final StringStreamParser instance = new StringStreamParser();
-
- private StringStreamParser() {
- }
-
- @Override
- public ParsedStreamMessage parse(StreamMessage streamMessage) {
- return new ParsedStreamMessage(Lists.newArrayList(new String(streamMessage.getRawData()).split(",")), streamMessage.getOffset(), streamMessage.getOffset(), true);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/TimePeriodCondition.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/TimePeriodCondition.java b/streaming/src/main/java/org/apache/kylin/streaming/TimePeriodCondition.java
deleted file mode 100644
index 560504c..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/TimePeriodCondition.java
+++ /dev/null
@@ -1,43 +0,0 @@
-package org.apache.kylin.streaming;
-
-/**
- */
-public class TimePeriodCondition implements BatchCondition {
-
- private final long startTime;
- private final long endTime;
- private final long margin;
-
- public TimePeriodCondition(long startTime, long endTime) {
- this(startTime, endTime, 0L);
- }
-
- public TimePeriodCondition(long startTime, long endTime, long margin) {
- this.startTime = startTime;
- this.endTime = endTime;
- this.margin = margin;
- }
-
- public long getStartTime() {
- return startTime;
- }
-
- public long getEndTime() {
- return endTime;
- }
-
- @Override
- public Result apply(ParsedStreamMessage message) {
- final long timestamp = message.getTimestamp();
- if (timestamp < startTime) {
- return Result.DISCARD;
- } else if (timestamp < endTime) {
- return Result.ACCEPT;
- } else if (timestamp < endTime + margin) {
- return Result.DISCARD;
- } else {
- return Result.REJECT;
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/TimedJsonStreamParser.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/TimedJsonStreamParser.java b/streaming/src/main/java/org/apache/kylin/streaming/TimedJsonStreamParser.java
deleted file mode 100644
index 5712d6e..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/TimedJsonStreamParser.java
+++ /dev/null
@@ -1,139 +0,0 @@
-/*
- *
- *
- * 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.streaming;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.kylin.common.util.DateFormat;
-import org.apache.kylin.common.util.TimeUtil;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.fasterxml.jackson.databind.JavaType;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.type.MapType;
-import com.fasterxml.jackson.databind.type.SimpleType;
-import com.google.common.collect.Lists;
-
-/**
- * each json message with a "timestamp" field
- */
-public final class TimedJsonStreamParser implements StreamParser {
-
- private static final Logger logger = LoggerFactory.getLogger(TimedJsonStreamParser.class);
-
- private List<TblColRef> allColumns;
- private boolean formatTs = false;
- private final ObjectMapper mapper = new ObjectMapper();
- private String tsColName = "timestamp";
- private final JavaType mapType = MapType.construct(HashMap.class, SimpleType.construct(String.class), SimpleType.construct(String.class));
-
- public TimedJsonStreamParser(List<TblColRef> allColumns, String propertiesStr) {
- this.allColumns = allColumns;
- if (!StringUtils.isEmpty(propertiesStr)) {
- String[] properties = propertiesStr.split(";");
- for (String prop : properties) {
- try {
- String[] parts = prop.split("=");
- if (parts.length == 2) {
- switch (parts[0]) {
- case "formatTs":
- this.formatTs = Boolean.valueOf(parts[1]);
- break;
- case "tsColName":
- this.tsColName = parts[1];
- break;
- default:
- break;
- }
- }
- } catch (Exception e) {
- logger.error("Failed to parse property " + prop);
- //ignore
- }
- }
- }
-
- logger.info("TimedJsonStreamParser with formatTs {} tsColName {}", formatTs, tsColName);
- }
-
- @Override
- public ParsedStreamMessage parse(StreamMessage stream) {
- try {
- Map<String, String> root = mapper.readValue(stream.getRawData(), mapType);
- String tsStr = root.get(tsColName);
- //Preconditions.checkArgument(!StringUtils.isEmpty(tsStr), "Timestamp field " + tsColName + //
- //" cannot be null, the message offset is " + stream.getOffset() + " content is " + new String(stream.getRawData()));
- long t;
- if (StringUtils.isEmpty(tsStr)) {
- t = 0;
- } else {
- t = Long.valueOf(tsStr);
- }
- ArrayList<String> result = Lists.newArrayList();
-
- for (TblColRef column : allColumns) {
- String columnName = column.getName();
- if (columnName.equalsIgnoreCase("minute_start")) {
- long minuteStart = TimeUtil.getMinuteStart(t);
- result.add(formatTs ? DateFormat.formatToTimeStr(minuteStart) : String.valueOf(minuteStart));
- } else if (columnName.equalsIgnoreCase("hour_start")) {
- long hourStart = TimeUtil.getHourStart(t);
- result.add(formatTs ? DateFormat.formatToTimeStr(hourStart) : String.valueOf(hourStart));
- } else if (columnName.equalsIgnoreCase("day_start")) {
- //of day start we'll add yyyy-mm-dd
- long ts = TimeUtil.getDayStart(t);
- result.add(DateFormat.formatToDateStr(ts));
- } else {
- String x = root.get(columnName.toLowerCase());
- result.add(x);
- }
- }
-
- return new ParsedStreamMessage(result, stream.getOffset(), t, true);
-
- } catch (IOException e) {
- logger.error("error", e);
- throw new RuntimeException(e);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/TopicMeta.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/TopicMeta.java b/streaming/src/main/java/org/apache/kylin/streaming/TopicMeta.java
deleted file mode 100644
index 4338b18..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/TopicMeta.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- *
- *
- * 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.streaming;
-
-import java.util.Collections;
-import java.util.List;
-
-/**
- * The topic metadata should be invariant, otherwise will cause re-initialization of the Consumer
- *
- */
-public class TopicMeta {
-
- private final String name;
-
- private final List<Integer> partitionIds;
-
- public TopicMeta(String name, List<Integer> partitionIds) {
- this.name = name;
- this.partitionIds = Collections.unmodifiableList(partitionIds);
- }
-
- public String getName() {
- return name;
- }
-
- public List<Integer> getPartitionIds() {
- return partitionIds;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamConsumer.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamConsumer.java b/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamConsumer.java
deleted file mode 100644
index 2a9e3e9..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamConsumer.java
+++ /dev/null
@@ -1,149 +0,0 @@
-/*
- *
- *
- * 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.streaming.invertedindex;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.invertedindex.index.Slice;
-import org.apache.kylin.invertedindex.model.IIDesc;
-import org.apache.kylin.invertedindex.model.IIKeyValueCodec;
-import org.apache.kylin.invertedindex.model.IIRow;
-import org.apache.kylin.storage.hbase.HBaseConnection;
-import org.apache.kylin.streaming.MicroStreamBatch;
-import org.apache.kylin.streaming.MicroStreamBatchConsumer;
-import org.apache.kylin.streaming.StreamingManager;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Stopwatch;
-import com.google.common.collect.Lists;
-
-/**
- */
-public class IIStreamConsumer implements MicroStreamBatchConsumer {
-
- private static Logger logger = LoggerFactory.getLogger(IIStreamConsumer.class);
-
- private final IIDesc desc;
- private final HTableInterface hTable;
- private final SliceBuilder sliceBuilder;
- private final int shardId;
- private final String streaming;
- private StreamingManager streamingManager;
-
- public IIStreamConsumer(String streaming, String hTableName, IIDesc iiDesc, int shard) {
- this.streaming = streaming;
- this.desc = iiDesc;
- this.shardId = shard;
- try {
- this.hTable = HBaseConnection.get(KylinConfig.getInstanceFromEnv().getStorageUrl()).getTable(hTableName);
- this.hTable.setAutoFlushTo(true);
- } catch (IOException e) {
- logger.error("cannot open htable name:" + hTableName, e);
- throw new RuntimeException("cannot open htable name:" + hTableName, e);
- }
- this.sliceBuilder = new SliceBuilder(desc, (short) shard, iiDesc.isUseLocalDictionary());
- this.streamingManager = StreamingManager.getInstance(KylinConfig.getInstanceFromEnv());
- }
-
- @Override
- public void consume(MicroStreamBatch microStreamBatch) throws IOException {
- if (microStreamBatch.size() > 0) {
- long offset = microStreamBatch.getOffset().getFirst();
- if (offset < streamingManager.getOffset(streaming, shardId)) {
- logger.info("this batch has already been built, skip building");
- return;
- }
- logger.info("stream build start, size:" + microStreamBatch.size());
- Stopwatch stopwatch = new Stopwatch();
- stopwatch.start();
- final Slice slice = sliceBuilder.buildSlice(microStreamBatch);
- logger.info("slice info, shard:" + slice.getShard() + " timestamp:" + slice.getTimestamp() + " record count:" + slice.getRecordCount());
-
- loadToHBase(hTable, slice, new IIKeyValueCodec(slice.getInfo()));
- submitOffset(offset);
- stopwatch.stop();
- logger.info("stream build finished, size:" + microStreamBatch.size() + " elapsed time:" + stopwatch.elapsedTime(TimeUnit.MILLISECONDS) + " " + TimeUnit.MILLISECONDS);
- } else {
- logger.info("nothing to build, skip building");
- }
- }
-
- @Override
- public void stop() {
- try {
- this.hTable.close();
- } catch (IOException e) {
- logger.error("onStop throw exception", e);
- }
- }
-
- private void loadToHBase(HTableInterface hTable, Slice slice, IIKeyValueCodec codec) throws IOException {
- List<Put> data = Lists.newArrayList();
- for (IIRow row : codec.encodeKeyValue(slice)) {
- final byte[] key = row.getKey().get();
- final byte[] value = row.getValue().get();
- Put put = new Put(key);
- put.add(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_QUALIFIER_BYTES, value);
- final ImmutableBytesWritable dictionary = row.getDictionary();
- final byte[] dictBytes = dictionary.get();
- if (dictionary.getOffset() == 0 && dictionary.getLength() == dictBytes.length) {
- put.add(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_DICTIONARY_BYTES, dictBytes);
- } else {
- throw new RuntimeException("dict offset should be 0, and dict length should be " + dictBytes.length + " but they are" + dictionary.getOffset() + " " + dictionary.getLength());
- }
- data.add(put);
- }
- hTable.put(data);
- //omit hTable.flushCommits(), because htable is auotflush
- }
-
- private void submitOffset(long offset) {
- try {
- streamingManager.updateOffset(streaming, shardId, offset);
- logger.info("submit offset:" + offset);
- } catch (Exception e) {
- logger.warn("error submit offset: " + offset + " retrying", e);
- throw new RuntimeException(e);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/15e1752e/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/SliceBuilder.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/SliceBuilder.java b/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/SliceBuilder.java
deleted file mode 100644
index 0fd0110..0000000
--- a/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/SliceBuilder.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * 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.streaming.invertedindex;
-
-import java.util.List;
-
-import javax.annotation.Nullable;
-
-import org.apache.kylin.dict.Dictionary;
-import org.apache.kylin.invertedindex.index.BatchSliceMaker;
-import org.apache.kylin.invertedindex.index.Slice;
-import org.apache.kylin.invertedindex.index.TableRecord;
-import org.apache.kylin.invertedindex.index.TableRecordInfo;
-import org.apache.kylin.invertedindex.model.IIDesc;
-import org.apache.kylin.invertedindex.util.IIDictionaryBuilder;
-import org.apache.kylin.streaming.MicroStreamBatch;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Function;
-import com.google.common.collect.Lists;
-
-/**
- */
-public final class SliceBuilder {
-
- private static Logger logger = LoggerFactory.getLogger(SliceBuilder.class);
-
- private final BatchSliceMaker sliceMaker;
- private final IIDesc iiDesc;
- private final boolean useLocalDict;
-
- public SliceBuilder(IIDesc desc, short shard, boolean useLocalDict) {
- this.iiDesc = desc;
- this.sliceMaker = new BatchSliceMaker(desc, shard);
- this.useLocalDict = useLocalDict;
- }
-
- public Slice buildSlice(MicroStreamBatch microStreamBatch) {
- final Dictionary<?>[] dictionaries = useLocalDict ? IIDictionaryBuilder.buildDictionary(microStreamBatch.getStreams(), iiDesc) : new Dictionary[iiDesc.listAllColumns().size()];
- TableRecordInfo tableRecordInfo = new TableRecordInfo(iiDesc, dictionaries);
- return build(microStreamBatch.getStreams(), tableRecordInfo, dictionaries);
- }
-
- private Slice build(List<List<String>> table, final TableRecordInfo tableRecordInfo, Dictionary<?>[] localDictionary) {
- final Slice slice = sliceMaker.makeSlice(tableRecordInfo.getDigest(), Lists.transform(table, new Function<List<String>, TableRecord>() {
- @Nullable
- @Override
- public TableRecord apply(@Nullable List<String> input) {
- TableRecord result = tableRecordInfo.createTableRecord();
- for (int i = 0; i < input.size(); i++) {
- result.setValueString(i, input.get(i));
- }
- return result;
- }
- }));
- slice.setLocalDictionaries(localDictionary);
- return slice;
- }
-}