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 2017/05/19 23:58:08 UTC
[30/50] [abbrv] kylin git commit: KYLIN-2596 Enable generating
multiple streaming messages with one input message in streaming parser
KYLIN-2596 Enable generating multiple streaming messages with one input message in streaming parser
* Minor, remove useless imports.
* Enable generating multiple streaming messages with one input message in streaming parser
* Make MR input can generate multiple rows of date.
* For multiple rows, outputKV() should be called for each row.
* Try&catch for each row's data processing.
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/edc4d4cc
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/edc4d4cc
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/edc4d4cc
Branch: refs/heads/KYLIN-2624
Commit: edc4d4cc558473476b18d48f232635e44640c27a
Parents: 7c0038d
Author: nichunen <zj...@sjtu.org>
Authored: Fri May 12 14:55:14 2017 +0800
Committer: hongbin ma <ma...@kyligence.io>
Committed: Fri May 12 14:55:14 2017 +0800
----------------------------------------------------------------------
.../java/org/apache/kylin/job/DeployUtil.java | 2 +-
.../kylin/common/util/StreamingMessage.java | 62 ----------
.../kylin/common/util/StreamingMessageRow.java | 62 ++++++++++
.../org/apache/kylin/engine/mr/IMRInput.java | 4 +-
.../mr/steps/FactDistinctColumnsMapper.java | 113 ++++++++++---------
.../engine/mr/steps/HiveToBaseCuboidMapper.java | 15 ++-
.../engine/mr/steps/InMemCuboidMapper.java | 15 ++-
.../apache/kylin/source/hive/HiveMRInput.java | 6 +-
.../cardinality/ColumnCardinalityMapper.java | 25 ++--
.../apache/kylin/source/kafka/KafkaMRInput.java | 16 ++-
.../kylin/source/kafka/StreamingParser.java | 8 +-
.../source/kafka/TimedJsonStreamParser.java | 11 +-
.../source/kafka/TimedJsonStreamParserTest.java | 22 ++--
13 files changed, 194 insertions(+), 167 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/assembly/src/test/java/org/apache/kylin/job/DeployUtil.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/job/DeployUtil.java b/assembly/src/test/java/org/apache/kylin/job/DeployUtil.java
index fdcd52c..077c056 100644
--- a/assembly/src/test/java/org/apache/kylin/job/DeployUtil.java
+++ b/assembly/src/test/java/org/apache/kylin/job/DeployUtil.java
@@ -166,7 +166,7 @@ public class DeployUtil {
TimedJsonStreamParser timedJsonStreamParser = new TimedJsonStreamParser(tableColumns, null);
StringBuilder sb = new StringBuilder();
for (String json : data) {
- List<String> rowColumns = timedJsonStreamParser.parse(ByteBuffer.wrap(json.getBytes())).getData();
+ List<String> rowColumns = timedJsonStreamParser.parse(ByteBuffer.wrap(json.getBytes())).get(0).getData();
sb.append(StringUtils.join(rowColumns, ","));
sb.append(System.getProperty("line.separator"));
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessage.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessage.java b/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessage.java
deleted file mode 100644
index 981c8a8..0000000
--- a/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessage.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.common.util;
-
-import java.util.List;
-import java.util.Map;
-
-/**
- */
-public class StreamingMessage {
-
- private final List<String> data;
-
- private long offset;
-
- private long timestamp;
-
- private Map<String, Object> params;
-
- public StreamingMessage(List<String> data, long offset, long timestamp, Map<String, Object> params) {
- this.data = data;
- this.offset = offset;
- this.timestamp = timestamp;
- this.params = params;
- }
-
- public final List<String> getData() {
- return data;
- }
-
- public final long getOffset() {
- return offset;
- }
-
- public void setOffset(long offset) {
- this.offset = offset;
- }
-
- public final long getTimestamp() {
- return timestamp;
- }
-
- public Map<String, Object> getParams() {
- return params;
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessageRow.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessageRow.java b/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessageRow.java
new file mode 100644
index 0000000..9b287d4
--- /dev/null
+++ b/core-common/src/main/java/org/apache/kylin/common/util/StreamingMessageRow.java
@@ -0,0 +1,62 @@
+/*
+ * 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.common.util;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ */
+public class StreamingMessageRow {
+
+ private final List<String> data;
+
+ private long offset;
+
+ private long timestamp;
+
+ private Map<String, Object> params;
+
+ public StreamingMessageRow(List<String> data, long offset, long timestamp, Map<String, Object> params) {
+ this.data = data;
+ this.offset = offset;
+ this.timestamp = timestamp;
+ this.params = params;
+ }
+
+ public final List<String> getData() {
+ return data;
+ }
+
+ public final long getOffset() {
+ return offset;
+ }
+
+ public void setOffset(long offset) {
+ this.offset = offset;
+ }
+
+ public final long getTimestamp() {
+ return timestamp;
+ }
+
+ public Map<String, Object> getParams() {
+ return params;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/engine-mr/src/main/java/org/apache/kylin/engine/mr/IMRInput.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/IMRInput.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/IMRInput.java
index 10d4879..6b0e557 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/IMRInput.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/IMRInput.java
@@ -24,6 +24,8 @@ import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
import org.apache.kylin.metadata.model.ISegment;
import org.apache.kylin.metadata.model.TableDesc;
+import java.util.Collection;
+
/**
* Any ITableSource that wishes to serve as input of MapReduce build engine must adapt to this interface.
*/
@@ -50,7 +52,7 @@ public interface IMRInput {
public void configureJob(Job job);
/** Parse a mapper input object into column values. */
- public String[] parseMapperInput(Object mapperInput);
+ public Collection<String[]> parseMapperInput(Object mapperInput);
}
/**
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapper.java
index d36ae18..713b7f7 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsMapper.java
@@ -44,6 +44,7 @@ import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
+
/**
*/
public class FactDistinctColumnsMapper<KEYIN> extends FactDistinctColumnsMapperBase<KEYIN, Object> {
@@ -157,70 +158,72 @@ public class FactDistinctColumnsMapper<KEYIN> extends FactDistinctColumnsMapperB
@Override
public void doMap(KEYIN key, Object record, Context context) throws IOException, InterruptedException {
- String[] row = flatTableInputFormat.parseMapperInput(record);
-
- context.getCounter(RawDataCounter.BYTES).increment(countSizeInBytes(row));
- for (int i = 0; i < factDictCols.size(); i++) {
- String fieldValue = row[dictionaryColumnIndex[i]];
- if (fieldValue == null)
- continue;
-
- int reducerIndex;
- if (uhcIndex[i] == 0) {
- //for the normal dictionary column
- reducerIndex = columnIndexToReducerBeginId.get(i);
- } else {
- //for the uhc
- reducerIndex = columnIndexToReducerBeginId.get(i) + (fieldValue.hashCode() & 0x7fffffff) % uhcReducerCount;
- }
+ Collection<String[]> rowCollection = flatTableInputFormat.parseMapperInput(record);
+
+ for (String[] row: rowCollection) {
+ context.getCounter(RawDataCounter.BYTES).increment(countSizeInBytes(row));
+ for (int i = 0; i < factDictCols.size(); i++) {
+ String fieldValue = row[dictionaryColumnIndex[i]];
+ if (fieldValue == null)
+ continue;
+
+ int reducerIndex;
+ if (uhcIndex[i] == 0) {
+ //for the normal dictionary column
+ reducerIndex = columnIndexToReducerBeginId.get(i);
+ } else {
+ //for the uhc
+ reducerIndex = columnIndexToReducerBeginId.get(i) + (fieldValue.hashCode() & 0x7fffffff) % uhcReducerCount;
+ }
- tmpbuf.clear();
- byte[] valueBytes = Bytes.toBytes(fieldValue);
- int size = valueBytes.length + 1;
- if (size >= tmpbuf.capacity()) {
- tmpbuf = ByteBuffer.allocate(countNewSize(tmpbuf.capacity(), size));
- }
- tmpbuf.put(Bytes.toBytes(reducerIndex)[3]);
- tmpbuf.put(valueBytes);
- outputKey.set(tmpbuf.array(), 0, tmpbuf.position());
- DataType type = factDictCols.get(i).getType();
- sortableKey.init(outputKey, type);
- //judge type
- context.write(sortableKey, EMPTY_TEXT);
-
- // log a few rows for troubleshooting
- if (rowCount < 10) {
- logger.info("Sample output: " + factDictCols.get(i) + " '" + fieldValue + "' => reducer " + reducerIndex);
+ tmpbuf.clear();
+ byte[] valueBytes = Bytes.toBytes(fieldValue);
+ int size = valueBytes.length + 1;
+ if (size >= tmpbuf.capacity()) {
+ tmpbuf = ByteBuffer.allocate(countNewSize(tmpbuf.capacity(), size));
+ }
+ tmpbuf.put(Bytes.toBytes(reducerIndex)[3]);
+ tmpbuf.put(valueBytes);
+ outputKey.set(tmpbuf.array(), 0, tmpbuf.position());
+ DataType type = factDictCols.get(i).getType();
+ sortableKey.init(outputKey, type);
+ //judge type
+ context.write(sortableKey, EMPTY_TEXT);
+
+ // log a few rows for troubleshooting
+ if (rowCount < 10) {
+ logger.info("Sample output: " + factDictCols.get(i) + " '" + fieldValue + "' => reducer " + reducerIndex);
+ }
}
- }
- if (collectStatistics) {
- if (rowCount % 100 < samplingPercentage) {
- if (isUsePutRowKeyToHllNewAlgorithm) {
- putRowKeyToHLLNew(row);
- } else {
- putRowKeyToHLLOld(row);
+ if (collectStatistics) {
+ if (rowCount % 100 < samplingPercentage) {
+ if (isUsePutRowKeyToHllNewAlgorithm) {
+ putRowKeyToHLLNew(row);
+ } else {
+ putRowKeyToHLLOld(row);
+ }
}
- }
- if (needFetchPartitionCol == true) {
- String fieldValue = row[partitionColumnIndex];
- if (fieldValue != null) {
- tmpbuf.clear();
- byte[] valueBytes = Bytes.toBytes(fieldValue);
- int size = valueBytes.length + 1;
- if (size >= tmpbuf.capacity()) {
- tmpbuf = ByteBuffer.allocate(countNewSize(tmpbuf.capacity(), size));
+ if (needFetchPartitionCol == true) {
+ String fieldValue = row[partitionColumnIndex];
+ if (fieldValue != null) {
+ tmpbuf.clear();
+ byte[] valueBytes = Bytes.toBytes(fieldValue);
+ int size = valueBytes.length + 1;
+ if (size >= tmpbuf.capacity()) {
+ tmpbuf = ByteBuffer.allocate(countNewSize(tmpbuf.capacity(), size));
+ }
+ tmpbuf.put(MARK_FOR_PARTITION_COL);
+ tmpbuf.put(valueBytes);
+ outputKey.set(tmpbuf.array(), 0, tmpbuf.position());
+ sortableKey.init(outputKey, (byte) 0);
+ context.write(sortableKey, EMPTY_TEXT);
}
- tmpbuf.put(MARK_FOR_PARTITION_COL);
- tmpbuf.put(valueBytes);
- outputKey.set(tmpbuf.array(), 0, tmpbuf.position());
- sortableKey.init(outputKey, (byte) 0);
- context.write(sortableKey, EMPTY_TEXT);
}
}
+ rowCount++;
}
- rowCount++;
}
private long countSizeInBytes(String[] row) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/HiveToBaseCuboidMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/HiveToBaseCuboidMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/HiveToBaseCuboidMapper.java
index 428f878..a04fb43 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/HiveToBaseCuboidMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/HiveToBaseCuboidMapper.java
@@ -19,6 +19,7 @@
package org.apache.kylin.engine.mr.steps;
import java.io.IOException;
+import java.util.Collection;
import org.apache.kylin.engine.mr.IMRInput.IMRTableInputFormat;
import org.apache.kylin.engine.mr.MRUtil;
@@ -38,12 +39,14 @@ public class HiveToBaseCuboidMapper<KEYIN> extends BaseCuboidMapperBase<KEYIN, O
@Override
public void doMap(KEYIN key, Object value, Context context) throws IOException, InterruptedException {
- String[] row = flatTableInputFormat.parseMapperInput(value);
- try {
- outputKV(row, context);
-
- } catch (Exception ex) {
- handleErrorRecord(row, ex);
+ Collection<String[]> rowCollection = flatTableInputFormat.parseMapperInput(value);
+ for (String[] row: rowCollection) {
+ try {
+ outputKV(row, context);
+
+ } catch (Exception ex) {
+ handleErrorRecord(row, ex);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidMapper.java
index c0ff2f2..eee189c 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidMapper.java
@@ -20,6 +20,7 @@ package org.apache.kylin.engine.mr.steps;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -116,12 +117,14 @@ public class InMemCuboidMapper<KEYIN> extends KylinMapper<KEYIN, Object, ByteArr
@Override
public void doMap(KEYIN key, Object record, Context context) throws IOException, InterruptedException {
// put each row to the queue
- String[] row = flatTableInputFormat.parseMapperInput(record);
- List<String> rowAsList = Arrays.asList(row);
-
- while (!future.isDone()) {
- if (queue.offer(rowAsList, 1, TimeUnit.SECONDS)) {
- break;
+ Collection<String[]> rowCollection = flatTableInputFormat.parseMapperInput(record);
+
+ for(String[] row: rowCollection) {
+ List<String> rowAsList = Arrays.asList(row);
+ while (!future.isDone()) {
+ if (queue.offer(rowAsList, 1, TimeUnit.SECONDS)) {
+ break;
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
----------------------------------------------------------------------
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
index 2f348a0..d7a2c7e 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
@@ -19,6 +19,8 @@
package org.apache.kylin.source.hive;
import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
import java.util.Set;
import org.apache.commons.lang.StringUtils;
@@ -118,8 +120,8 @@ public class HiveMRInput implements IMRInput {
}
@Override
- public String[] parseMapperInput(Object mapperInput) {
- return HiveTableReader.getRowAsStringArray((HCatRecord) mapperInput);
+ public List<String[]> parseMapperInput(Object mapperInput) {
+ return Collections.singletonList(HiveTableReader.getRowAsStringArray((HCatRecord) mapperInput));
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
----------------------------------------------------------------------
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java b/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
index c712605..9033d67 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
@@ -20,6 +20,7 @@ package org.apache.kylin.source.hive.cardinality;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
@@ -68,22 +69,24 @@ public class ColumnCardinalityMapper<T> extends KylinMapper<T, Object, IntWritab
@Override
public void doMap(T key, Object value, Context context) throws IOException, InterruptedException {
ColumnDesc[] columns = tableDesc.getColumns();
- String[] values = tableInputFormat.parseMapperInput(value);
+ Collection<String[]> valuesCollection = tableInputFormat.parseMapperInput(value);
- for (int m = 0; m < columns.length; m++) {
- String field = columns[m].getName();
- String fieldValue = values[m];
- if (fieldValue == null)
- fieldValue = "NULL";
+ for (String[] values: valuesCollection) {
+ for (int m = 0; m < columns.length; m++) {
+ String field = columns[m].getName();
+ String fieldValue = values[m];
+ if (fieldValue == null)
+ fieldValue = "NULL";
- if (counter < 5 && m < 10) {
- System.out.println("Get row " + counter + " column '" + field + "' value: " + fieldValue);
+ if (counter < 5 && m < 10) {
+ System.out.println("Get row " + counter + " column '" + field + "' value: " + fieldValue);
+ }
+
+ getHllc(m).add(Bytes.toBytes(fieldValue.toString()));
}
- getHllc(m).add(Bytes.toBytes(fieldValue.toString()));
+ counter++;
}
-
- counter++;
}
private HLLCounter getHllc(Integer key) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
----------------------------------------------------------------------
diff --git a/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
index c7b327f..500e1e9 100644
--- a/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
+++ b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
@@ -19,7 +19,9 @@ package org.apache.kylin.source.kafka;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.List;
import javax.annotation.Nullable;
@@ -32,7 +34,7 @@ import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.HadoopUtil;
-import org.apache.kylin.common.util.StreamingMessage;
+import org.apache.kylin.common.util.StreamingMessageRow;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.model.CubeJoinedFlatTableDesc;
import org.apache.kylin.engine.mr.IMRInput;
@@ -127,11 +129,17 @@ public class KafkaMRInput implements IMRInput {
}
@Override
- public String[] parseMapperInput(Object mapperInput) {
+ public Collection<String[]> parseMapperInput(Object mapperInput) {
Text text = (Text) mapperInput;
ByteBuffer buffer = ByteBuffer.wrap(text.getBytes(), 0, text.getLength());
- StreamingMessage streamingMessage = streamingParser.parse(buffer);
- return streamingMessage.getData().toArray(new String[streamingMessage.getData().size()]);
+ List<StreamingMessageRow> streamingMessageRowList = streamingParser.parse(buffer);
+ List<String[]> parsedDataCollection = new ArrayList<>();
+
+ for (StreamingMessageRow row: streamingMessageRowList) {
+ parsedDataCollection.add(row.getData().toArray(new String[row.getData().size()]));
+ }
+
+ return parsedDataCollection;
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/source-kafka/src/main/java/org/apache/kylin/source/kafka/StreamingParser.java
----------------------------------------------------------------------
diff --git a/source-kafka/src/main/java/org/apache/kylin/source/kafka/StreamingParser.java b/source-kafka/src/main/java/org/apache/kylin/source/kafka/StreamingParser.java
index 75f9c4b..2e3c11c 100644
--- a/source-kafka/src/main/java/org/apache/kylin/source/kafka/StreamingParser.java
+++ b/source-kafka/src/main/java/org/apache/kylin/source/kafka/StreamingParser.java
@@ -26,7 +26,7 @@ import com.google.common.collect.Maps;
import org.apache.commons.lang3.StringUtils;
import java.nio.ByteBuffer;
import org.apache.kylin.common.util.DateFormat;
-import org.apache.kylin.common.util.StreamingMessage;
+import org.apache.kylin.common.util.StreamingMessageRow;
import org.apache.kylin.common.util.TimeUtil;
import org.apache.kylin.metadata.model.TblColRef;
@@ -62,11 +62,11 @@ public abstract class StreamingParser {
/**
* @param message
- * @return StreamingMessage must not be NULL
+ * @return List<StreamingMessageRow> must not be NULL
*/
- abstract public StreamingMessage parse(ByteBuffer message);
+ abstract public List<StreamingMessageRow> parse(ByteBuffer message);
- abstract public boolean filter(StreamingMessage streamingMessage);
+ abstract public boolean filter(StreamingMessageRow streamingMessageRow);
public static StreamingParser getStreamingParser(String parserName, String parserProperties, List<TblColRef> columns) throws ReflectiveOperationException {
if (!StringUtils.isEmpty(parserName)) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/source-kafka/src/main/java/org/apache/kylin/source/kafka/TimedJsonStreamParser.java
----------------------------------------------------------------------
diff --git a/source-kafka/src/main/java/org/apache/kylin/source/kafka/TimedJsonStreamParser.java b/source-kafka/src/main/java/org/apache/kylin/source/kafka/TimedJsonStreamParser.java
index 6ff0d2f..de167b4 100644
--- a/source-kafka/src/main/java/org/apache/kylin/source/kafka/TimedJsonStreamParser.java
+++ b/source-kafka/src/main/java/org/apache/kylin/source/kafka/TimedJsonStreamParser.java
@@ -32,7 +32,7 @@ import java.util.Arrays;
import com.fasterxml.jackson.databind.DeserializationFeature;
import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.common.util.ByteBufferBackedInputStream;
-import org.apache.kylin.common.util.StreamingMessage;
+import org.apache.kylin.common.util.StreamingMessageRow;
import org.apache.kylin.metadata.model.TblColRef;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -100,7 +100,7 @@ public final class TimedJsonStreamParser extends StreamingParser {
}
@Override
- public StreamingMessage parse(ByteBuffer buffer) {
+ public List<StreamingMessageRow> parse(ByteBuffer buffer) {
try {
Map<String, Object> message = mapper.readValue(new ByteBufferBackedInputStream(buffer), mapType);
root.clear();
@@ -116,7 +116,10 @@ public final class TimedJsonStreamParser extends StreamingParser {
}
}
- return new StreamingMessage(result, 0, t, Collections.<String, Object>emptyMap());
+ StreamingMessageRow streamingMessageRow = new StreamingMessageRow(result, 0, t, Collections.<String, Object>emptyMap());
+ List<StreamingMessageRow> messageRowList = new ArrayList<StreamingMessageRow>();
+ messageRowList.add(streamingMessageRow);
+ return messageRowList;
} catch (IOException e) {
logger.error("error", e);
throw new RuntimeException(e);
@@ -124,7 +127,7 @@ public final class TimedJsonStreamParser extends StreamingParser {
}
@Override
- public boolean filter(StreamingMessage streamingMessage) {
+ public boolean filter(StreamingMessageRow streamingMessageRow) {
return true;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/edc4d4cc/source-kafka/src/test/java/org/apache/kylin/source/kafka/TimedJsonStreamParserTest.java
----------------------------------------------------------------------
diff --git a/source-kafka/src/test/java/org/apache/kylin/source/kafka/TimedJsonStreamParserTest.java b/source-kafka/src/test/java/org/apache/kylin/source/kafka/TimedJsonStreamParserTest.java
index 230ff00..8dc840b 100644
--- a/source-kafka/src/test/java/org/apache/kylin/source/kafka/TimedJsonStreamParserTest.java
+++ b/source-kafka/src/test/java/org/apache/kylin/source/kafka/TimedJsonStreamParserTest.java
@@ -28,7 +28,7 @@ import java.util.List;
import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.common.util.LocalFileMetadataTestCase;
-import org.apache.kylin.common.util.StreamingMessage;
+import org.apache.kylin.common.util.StreamingMessageRow;
import org.apache.kylin.metadata.model.TableDesc;
import org.apache.kylin.metadata.model.TblColRef;
import org.junit.AfterClass;
@@ -65,8 +65,8 @@ public class TimedJsonStreamParserTest extends LocalFileMetadataTestCase {
TimedJsonStreamParser parser = new TimedJsonStreamParser(allCol, null);
Object msg = mapper.readValue(new File(jsonFilePath), mapType);
ByteBuffer buffer = getJsonByteBuffer(msg);
- StreamingMessage sMsg = parser.parse(buffer);
- List<String> result = sMsg.getData();
+ List<StreamingMessageRow> msgList = parser.parse(buffer);
+ List<String> result = msgList.get(0).getData();
assertEquals("Jul 20, 2016 9:59:17 AM", result.get(0));
assertEquals("755703618762862600", result.get(1));
assertEquals("false", result.get(2));
@@ -80,8 +80,8 @@ public class TimedJsonStreamParserTest extends LocalFileMetadataTestCase {
TimedJsonStreamParser parser = new TimedJsonStreamParser(allCol, null);
Object msg = mapper.readValue(new File(jsonFilePath), mapType);
ByteBuffer buffer = getJsonByteBuffer(msg);
- StreamingMessage sMsg = parser.parse(buffer);
- List<String> result = sMsg.getData();
+ List<StreamingMessageRow> msgList = parser.parse(buffer);
+ List<String> result = msgList.get(0).getData();
assertEquals("4853763947", result.get(0));
assertEquals("Noticias", result.get(1));
assertEquals("false", result.get(2));
@@ -96,8 +96,8 @@ public class TimedJsonStreamParserTest extends LocalFileMetadataTestCase {
HashMap<String, Object> map = (HashMap<String, Object>) msg;
Object array = map.get("mediaEntities");
ByteBuffer buffer = getJsonByteBuffer(msg);
- StreamingMessage sMsg = parser.parse(buffer);
- List<String> result = sMsg.getData();
+ List<StreamingMessageRow> msgList = parser.parse(buffer);
+ List<String> result = msgList.get(0).getData();
System.out.println(result);
}
@@ -109,8 +109,8 @@ public class TimedJsonStreamParserTest extends LocalFileMetadataTestCase {
TimedJsonStreamParser parser = new TimedJsonStreamParser(allCol, null);
Object msg = mapper.readValue(new File(jsonFilePath), mapType);
ByteBuffer buffer = getJsonByteBuffer(msg);
- StreamingMessage sMsg = parser.parse(buffer);
- List<String> result = sMsg.getData();
+ List<StreamingMessageRow> msgList = parser.parse(buffer);
+ List<String> result = msgList.get(0).getData();
}
@@ -121,8 +121,8 @@ public class TimedJsonStreamParserTest extends LocalFileMetadataTestCase {
TimedJsonStreamParser parser = new TimedJsonStreamParser(allCol, null);
Object msg = mapper.readValue(new File(jsonFilePath), mapType);
ByteBuffer buffer = getJsonByteBuffer(msg);
- StreamingMessage sMsg = parser.parse(buffer);
- List<String> result = sMsg.getData();
+ List<StreamingMessageRow> msgList = parser.parse(buffer);
+ List<String> result = msgList.get(0).getData();
assertEquals(StringUtils.EMPTY, result.get(0));
assertEquals(StringUtils.EMPTY, result.get(1));
}