You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by nj...@apache.org on 2017/08/15 16:56:50 UTC
[2/7] kylin git commit: APACHE-KYLIN-2725: Introduce a tool for
creating system cubes relating to query & job metrics
http://git-wip-us.apache.org/repos/asf/kylin/blob/74e167f0/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/SCCreator.java
----------------------------------------------------------------------
diff --git a/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/SCCreator.java b/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/SCCreator.java
new file mode 100644
index 0000000..fb2fbb5
--- /dev/null
+++ b/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/SCCreator.java
@@ -0,0 +1,292 @@
+/*
+ * 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.tool.metrics.systemcube;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedWriter;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.OptionBuilder;
+import org.apache.commons.cli.Options;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.persistence.RootPersistentEntity;
+import org.apache.kylin.common.persistence.Serializer;
+import org.apache.kylin.common.util.AbstractApplication;
+import org.apache.kylin.common.util.OptionsHelper;
+import org.apache.kylin.cube.CubeDescManager;
+import org.apache.kylin.cube.CubeInstance;
+import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.cube.model.CubeDesc;
+import org.apache.kylin.metadata.MetadataManager;
+import org.apache.kylin.metadata.model.DataModelDesc;
+import org.apache.kylin.metadata.model.TableDesc;
+import org.apache.kylin.metadata.project.ProjectInstance;
+import org.apache.kylin.metrics.lib.SinkTool;
+import org.apache.kylin.tool.metrics.systemcube.util.HiveSinkTool;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
+
+public class SCCreator extends AbstractApplication {
+
+ private static final Logger logger = LoggerFactory.getLogger(SCCreator.class);
+
+ private static final Option OPTION_OWNER = OptionBuilder.withArgName("owner").hasArg().isRequired(false)
+ .withDescription("Specify the owner who creates the metadata").create("owner");
+ private static final Option OPTION_INPUT_CONFIG = OptionBuilder.withArgName("inputConfig").hasArg()
+ .isRequired(false).withDescription("Specify the input configuration file").create("inputConfig");
+ private static final Option OPTION_OUTPUT = OptionBuilder.withArgName("output").hasArg().isRequired(true)
+ .withDescription("Specify the output where the generated metadata will be saved").create("output");
+
+ private static final String D_CUBE_INSTANCE = "cube/";
+ private static final String D_CUBE_DESC = "cube_desc/";
+ private static final String D_PROJECT = "project/";
+ private static final String D_TABLE = "table/";
+ private static final String D_MODEL_DESC = "model_desc/";
+ // private static final String D_STREAMING = "streaming/";
+
+ private static final String F_HIVE_SQL = "create_hive_tables_for_system_cubes";
+
+ protected final Options options;
+
+ private final KylinConfig config;
+
+ private final ObjectMapper mapper;
+
+ public SCCreator() {
+ mapper = new ObjectMapper();
+ mapper.enableDefaultTyping();
+
+ config = KylinConfig.getInstanceFromEnv();
+
+ options = new Options();
+ options.addOption(OPTION_OWNER);
+ options.addOption(OPTION_OUTPUT);
+ options.addOption(OPTION_INPUT_CONFIG);
+ }
+
+ public static void main(String[] args) {
+ SCCreator cli = new SCCreator();
+ cli.execute(args);
+ }
+
+ protected Options getOptions() {
+ return options;
+ }
+
+ protected void execute(OptionsHelper optionsHelper) throws Exception {
+ String owner = optionsHelper.getOptionValue(OPTION_OWNER);
+ String output = optionsHelper.getOptionValue(OPTION_OUTPUT);
+ String inputConfig = optionsHelper.getOptionValue(OPTION_INPUT_CONFIG);
+ if (Strings.isNullOrEmpty(inputConfig)) {
+ throw new RuntimeException("Input configuration file should be specified!!!");
+ }
+
+ execute(owner, output, inputConfig);
+ }
+
+ public void execute(String owner, String output, String inputConfig) throws Exception {
+ if (Strings.isNullOrEmpty(owner)) {
+ owner = "ADMIN";
+ }
+ if (!output.endsWith("/")) {
+ output += "/";
+ }
+
+ Set<SinkTool> sourceToolSet = mapper
+ .readValue(new BufferedInputStream(new FileInputStream(new File(inputConfig))), HashSet.class);
+ run(owner, output, sourceToolSet);
+ }
+
+ private void run(String owner, String output, Collection<SinkTool> sinkToolSet) throws IOException {
+ List<TableDesc> kylinTables = Lists.newArrayList();
+ List<DataModelDesc> kylinModels = Lists.newArrayList();
+ List<CubeDesc> kylinCubeDescs = Lists.newArrayList();
+ List<CubeInstance> kylinCubeInstances = Lists.newArrayList();
+ // List<StreamingConfig> kylinStreamings = Lists.newArrayList();
+
+ boolean ifHive = false;
+ for (SinkTool sourceTool : sinkToolSet) {
+ if (sourceTool instanceof HiveSinkTool) {
+ ifHive = true;
+ } else {
+ logger.warn("current version only support hive sink!!!");
+ continue;
+ }
+ kylinTables.addAll(generateKylinTableForSystemCube(sourceTool));
+ kylinModels.addAll(generateKylinModelForSystemCube(owner, sourceTool));
+ kylinCubeDescs.addAll(generateKylinCubeDescForSystemCube(sourceTool));
+ kylinCubeInstances.addAll(generateKylinCubeInstanceForSystemCube(owner, sourceTool));
+ // if (sourceTool instanceof StreamSinkTool) {
+ // kylinStreamings.addAll(generateKylinStreamingForSystemCube((StreamSinkTool) sourceTool));
+ // }
+ }
+
+ if (ifHive) {
+ generateHiveTableSQLFileForSystemCube(output);
+ }
+
+ ProjectInstance projectInstance = ProjectCreator.generateKylinProjectInstance(owner, kylinTables, kylinModels,
+ kylinCubeDescs);
+ generateKylinProjectFileForSystemCube(output, projectInstance);
+ for (TableDesc tableDesc : kylinTables) {
+ generateKylinTableFileForSystemCube(output, tableDesc);
+ }
+ for (DataModelDesc dataModelDesc : kylinModels) {
+ generateKylinModelFileForSystemCube(output, dataModelDesc);
+ }
+ for (CubeDesc cubeDesc : kylinCubeDescs) {
+ generateKylinCubeDescFileForSystemCube(output, cubeDesc);
+ }
+ for (CubeInstance cubeInstance : kylinCubeInstances) {
+ generateKylinCubeInstanceFileForSystemCube(output, cubeInstance);
+ }
+ // for (StreamingConfig streaming : kylinStreamings) {
+ // generateKylinStreamingFileForSystemCube(output, streaming);
+ // }
+ }
+
+ private List<TableDesc> generateKylinTableForSystemCube(SinkTool sinkTool) {
+ List<TableDesc> result = Lists.newLinkedList();
+ result.add(KylinTableCreator.generateKylinTableForMetricsQuery(config, sinkTool));
+ result.add(KylinTableCreator.generateKylinTableForMetricsQueryCube(config, sinkTool));
+ result.add(KylinTableCreator.generateKylinTableForMetricsQueryRPC(config, sinkTool));
+ result.add(KylinTableCreator.generateKylinTableForMetricsJob(config, sinkTool));
+ result.add(KylinTableCreator.generateKylinTableForMetricsJobException(config, sinkTool));
+
+ return result;
+ }
+
+ private List<DataModelDesc> generateKylinModelForSystemCube(String owner, SinkTool sinkTool) {
+ List<DataModelDesc> result = Lists.newLinkedList();
+ result.add(ModelCreator.generateKylinModelForMetricsQuery(owner, config, sinkTool));
+ result.add(ModelCreator.generateKylinModelForMetricsQueryCube(owner, config, sinkTool));
+ result.add(ModelCreator.generateKylinModelForMetricsQueryRPC(owner, config, sinkTool));
+ result.add(ModelCreator.generateKylinModelForMetricsJob(owner, config, sinkTool));
+ result.add(ModelCreator.generateKylinModelForMetricsJobException(owner, config, sinkTool));
+
+ return result;
+ }
+
+ private List<CubeDesc> generateKylinCubeDescForSystemCube(SinkTool sinkTool) {
+ List<CubeDesc> result = Lists.newLinkedList();
+ result.add(CubeDescCreator.generateKylinCubeDescForMetricsQuery(config, sinkTool));
+ result.add(CubeDescCreator.generateKylinCubeDescForMetricsQueryCube(config, sinkTool));
+ result.add(CubeDescCreator.generateKylinCubeDescForMetricsQueryRPC(config, sinkTool));
+ result.add(CubeDescCreator.generateKylinCubeDescForMetricsJob(config, sinkTool));
+ result.add(CubeDescCreator.generateKylinCubeDescForMetricsJobException(config, sinkTool));
+
+ return result;
+ }
+
+ private List<CubeInstance> generateKylinCubeInstanceForSystemCube(String owner, SinkTool sinkTool) {
+ List<CubeInstance> result = Lists.newLinkedList();
+ result.add(CubeInstanceCreator.generateKylinCubeInstanceForMetricsQuery(owner, config, sinkTool));
+ result.add(CubeInstanceCreator.generateKylinCubeInstanceForMetricsQueryCube(owner, config, sinkTool));
+ result.add(CubeInstanceCreator.generateKylinCubeInstanceForMetricsQueryRPC(owner, config, sinkTool));
+ result.add(CubeInstanceCreator.generateKylinCubeInstanceForMetricsJob(owner, config, sinkTool));
+ result.add(CubeInstanceCreator.generateKylinCubeInstanceForMetricsJobException(owner, config, sinkTool));
+
+ return result;
+ }
+
+ // private List<StreamingConfig> generateKylinStreamingForSystemCube(StreamSinkTool sinkTool) {
+ // List<StreamingConfig> result = Lists.newLinkedList();
+ // result.add(StreamingCreator.generateKylinStreamingForMetricsQuery(config, sinkTool));
+ // result.add(StreamingCreator.generateKylinStreamingForMetricsQueryCube(config, sinkTool));
+ // result.add(StreamingCreator.generateKylinStreamingForMetricsQueryRPC(config, sinkTool));
+ // result.add(StreamingCreator.generateKylinStreamingForMetricsJob(config, sinkTool));
+ // result.add(StreamingCreator.generateKylinStreamingForMetricsJobException(config, sinkTool));
+ //
+ // return result;
+ // }
+
+ private void generateHiveTableSQLFileForSystemCube(String output) throws IOException {
+ String contents = HiveTableCreator.generateAllSQL(config);
+ saveToFile(output + F_HIVE_SQL + ".sql", contents);
+ }
+
+ private void generateKylinTableFileForSystemCube(String output, TableDesc kylinTable) throws IOException {
+ saveSystemCubeMetadataToFile(output + D_TABLE + kylinTable.getIdentity() + ".json", kylinTable,
+ MetadataManager.TABLE_SERIALIZER);
+ }
+
+ private void generateKylinModelFileForSystemCube(String output, DataModelDesc modelDesc) throws IOException {
+ saveSystemCubeMetadataToFile(output + D_MODEL_DESC + modelDesc.getName() + ".json", modelDesc,
+ MetadataManager.MODELDESC_SERIALIZER);
+ }
+
+ private void generateKylinCubeInstanceFileForSystemCube(String output, CubeInstance cubeInstance)
+ throws IOException {
+ saveSystemCubeMetadataToFile(output + D_CUBE_INSTANCE + cubeInstance.getName() + ".json", cubeInstance,
+ CubeManager.CUBE_SERIALIZER);
+ }
+
+ private void generateKylinCubeDescFileForSystemCube(String output, CubeDesc cubeDesc) throws IOException {
+ saveSystemCubeMetadataToFile(output + D_CUBE_DESC + cubeDesc.getName() + ".json", cubeDesc,
+ CubeDescManager.CUBE_DESC_SERIALIZER);
+ }
+
+ private void generateKylinProjectFileForSystemCube(String output, ProjectInstance projectInstance)
+ throws IOException {
+ saveSystemCubeMetadataToFile(output + D_PROJECT + projectInstance.getName() + ".json", projectInstance,
+ CubeDescManager.CUBE_DESC_SERIALIZER);
+ }
+
+ // private void generateKylinStreamingFileForSystemCube(String output, StreamingConfig streamingConfig)
+ // throws IOException {
+ // saveSystemCubeMetadataToFile(output + D_STREAMING + streamingConfig.getName() + ".json", streamingConfig,
+ // StreamingConfig.SERIALIZER);
+ // }
+
+ private <T extends RootPersistentEntity> void saveSystemCubeMetadataToFile(String fileName, T metadata,
+ Serializer serializer) throws IOException {
+ ByteArrayOutputStream buf = new ByteArrayOutputStream();
+ DataOutputStream dout = new DataOutputStream(buf);
+ serializer.serialize(metadata, dout);
+ dout.close();
+ buf.close();
+
+ saveToFile(fileName, buf.toString());
+ }
+
+ private void saveToFile(String fileName, String contents) throws IOException {
+ File parentDir = new File(fileName).getParentFile();
+ if (!parentDir.exists()) {
+ parentDir.mkdirs();
+ }
+
+ BufferedWriter bufferedWriter = new BufferedWriter(new FileWriter(fileName));
+ bufferedWriter.append(contents);
+ bufferedWriter.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/74e167f0/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/StreamingCreator.java
----------------------------------------------------------------------
diff --git a/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/StreamingCreator.java b/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/StreamingCreator.java
new file mode 100644
index 0000000..d5dfc94
--- /dev/null
+++ b/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/StreamingCreator.java
@@ -0,0 +1,138 @@
+/*
+ * 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.tool.metrics.systemcube;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.persistence.JsonSerializer;
+import org.apache.kylin.common.persistence.RootPersistentEntity;
+import org.apache.kylin.common.persistence.Serializer;
+import org.apache.kylin.tool.metrics.systemcube.util.KafkaSinkTool;
+import org.apache.kylin.tool.metrics.systemcube.util.StreamSinkTool;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.Maps;
+
+public class StreamingCreator {
+
+ public static void main(String[] args) throws Exception {
+ // KylinConfig.setSandboxEnvIfPossible();
+ KylinConfig config = KylinConfig.getInstanceFromEnv();
+
+ StreamingConfig streamingConfig = generateKylinStreamingForMetricsQuery(config, new KafkaSinkTool());
+ ByteArrayOutputStream buf = new ByteArrayOutputStream();
+ DataOutputStream dout = new DataOutputStream(buf);
+ StreamingConfig.SERIALIZER.serialize(streamingConfig, dout);
+ dout.close();
+ buf.close();
+ System.out.println(buf.toString());
+ }
+
+ public static StreamingConfig generateKylinStreamingForMetricsQuery(KylinConfig config, StreamSinkTool sinkTool) {
+ String tableName = sinkTool.getTableNameForMetrics(config.getKylinMetricsSubjectQuery());
+ return generateKylinStreaming(tableName, sinkTool.getTopicName(tableName), sinkTool.getStreamingType(),
+ sinkTool.getStreamingProperties());
+ }
+
+ public static StreamingConfig generateKylinStreamingForMetricsQueryCube(KylinConfig config,
+ StreamSinkTool sinkTool) {
+ String tableName = sinkTool.getTableNameForMetrics(config.getKylinMetricsSubjectQueryCube());
+ return generateKylinStreaming(tableName, sinkTool.getTopicName(tableName), sinkTool.getStreamingType(),
+ sinkTool.getStreamingProperties());
+ }
+
+ public static StreamingConfig generateKylinStreamingForMetricsQueryRPC(KylinConfig config,
+ StreamSinkTool sinkTool) {
+ String tableName = sinkTool.getTableNameForMetrics(config.getKylinMetricsSubjectQueryRpcCall());
+ return generateKylinStreaming(tableName, sinkTool.getTopicName(tableName), sinkTool.getStreamingType(),
+ sinkTool.getStreamingProperties());
+ }
+
+ public static StreamingConfig generateKylinStreamingForMetricsJob(KylinConfig config, StreamSinkTool sinkTool) {
+ String tableName = sinkTool.getTableNameForMetrics(config.getKylinMetricsSubjectJob());
+ return generateKylinStreaming(tableName, sinkTool.getTopicName(tableName), sinkTool.getStreamingType(),
+ sinkTool.getStreamingProperties());
+ }
+
+ public static StreamingConfig generateKylinStreamingForMetricsJobException(KylinConfig config,
+ StreamSinkTool sinkTool) {
+ String tableName = sinkTool.getTableNameForMetrics(config.getKylinMetricsSubjectJobException());
+ return generateKylinStreaming(tableName, sinkTool.getTopicName(tableName), sinkTool.getStreamingType(),
+ sinkTool.getStreamingProperties());
+ }
+
+ public static StreamingConfig generateKylinStreaming(String tableName, String topicName, String type,
+ Map<String, String> properties) {
+ StreamingConfig streamingConfig = new StreamingConfig();
+ streamingConfig.setUuid(UUID.randomUUID().toString());
+ streamingConfig.setName(tableName);
+ streamingConfig.setLastModified(0L);
+ streamingConfig.setType(type);
+
+ Map<String, String> streamingProperties = Maps.newHashMap(properties);
+ streamingProperties.put("topic", topicName);
+ streamingConfig.setProperties(streamingProperties);
+
+ return streamingConfig;
+ }
+
+ public static class StreamingConfig extends RootPersistentEntity {
+
+ public static final String STREAMING_TYPE_KAFKA = "kafka";
+ public static Serializer<StreamingConfig> SERIALIZER = new JsonSerializer<StreamingConfig>(
+ StreamingConfig.class);
+ @JsonProperty("name")
+ private String name;
+
+ @JsonProperty("type")
+ private String type = STREAMING_TYPE_KAFKA;
+
+ @JsonProperty("properties")
+ private Map<String, String> properties = Maps.newLinkedHashMap();
+
+ public String getType() {
+ return type;
+ }
+
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public Map<String, String> getProperties() {
+ return properties;
+ }
+
+ public void setProperties(Map<String, String> properties) {
+ this.properties = properties;
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/74e167f0/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/HiveSinkTool.java
----------------------------------------------------------------------
diff --git a/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/HiveSinkTool.java b/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/HiveSinkTool.java
new file mode 100644
index 0000000..ef46163
--- /dev/null
+++ b/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/HiveSinkTool.java
@@ -0,0 +1,61 @@
+/*
+ * 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.tool.metrics.systemcube.util;
+
+import java.util.Map;
+
+import org.apache.kylin.metadata.model.ISourceAware;
+import org.apache.kylin.metadata.model.IStorageAware;
+import org.apache.kylin.metrics.lib.SinkTool;
+import org.apache.kylin.metrics.lib.impl.hive.HiveReservoirReporter;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.Maps;
+
+@SuppressWarnings("serial")
+@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.NONE, getterVisibility = JsonAutoDetect.Visibility.NONE, isGetterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE)
+public class HiveSinkTool implements SinkTool {
+
+ @JsonProperty("storageType")
+ protected int storageType = IStorageAware.ID_SHARDED_HBASE;
+
+ @JsonProperty("cubeDescOverrideProperties")
+ protected Map<String, String> cubeDescOverrideProperties = Maps.newHashMap();
+
+ public int getStorageType() {
+ return storageType;
+ }
+
+ public int getSourceType() {
+ return ISourceAware.ID_HIVE;
+ }
+
+ public String getTableNameForMetrics(String subject) {
+ return HiveReservoirReporter.getTableFromSubject(subject);
+ }
+
+ public Map<String, String> getCubeDescOverrideProperties() {
+ return cubeDescOverrideProperties;
+ }
+
+ public void setCubeDescOverrideProperties(Map<String, String> cubeDescOverrideProperties) {
+ this.cubeDescOverrideProperties = cubeDescOverrideProperties;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/74e167f0/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/KafkaSinkTool.java
----------------------------------------------------------------------
diff --git a/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/KafkaSinkTool.java b/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/KafkaSinkTool.java
new file mode 100644
index 0000000..6caa5ec
--- /dev/null
+++ b/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/KafkaSinkTool.java
@@ -0,0 +1,32 @@
+/*
+ * 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.tool.metrics.systemcube.util;
+
+import org.apache.kylin.metrics.lib.impl.kafka.KafkaReservoirReporter;
+
+public class KafkaSinkTool extends StreamSinkTool {
+
+ public String getTableNameForMetrics(String subject) {
+ return KafkaReservoirReporter.getTableFromSubject(subject);
+ }
+
+ public String getStreamingType() {
+ return "kafka";
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/74e167f0/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/StreamSinkTool.java
----------------------------------------------------------------------
diff --git a/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/StreamSinkTool.java b/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/StreamSinkTool.java
new file mode 100644
index 0000000..2f6f034
--- /dev/null
+++ b/tool/src/main/java/org/apache/kylin/tool/metrics/systemcube/util/StreamSinkTool.java
@@ -0,0 +1,74 @@
+/*
+ * 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.tool.metrics.systemcube.util;
+
+import java.util.Map;
+
+import org.apache.kylin.metadata.model.ISourceAware;
+import org.apache.kylin.metadata.model.IStorageAware;
+import org.apache.kylin.metrics.lib.impl.RecordEvent;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.Maps;
+
+public abstract class StreamSinkTool {
+
+ @JsonProperty("cubeDescOverrideProperties")
+ protected Map<String, String> cubeDescOverrideProperties = Maps.newHashMap();
+
+ @JsonProperty("streamingProperties")
+ protected Map<String, String> streamingProperties = Maps.newHashMap();
+
+ public abstract String getTableNameForMetrics(String subject);
+
+ public int getStorageType() {
+ return IStorageAware.ID_STREAM;
+ }
+
+ public abstract String getStreamingType();
+
+ public String getTopicName(String tableName) {
+ return tableName.replace('.', '_');
+ }
+
+ public int getSourceType() {
+ return ISourceAware.ID_STREAMING;
+ }
+
+ public Map<String, String> getCubeDescOverrideProperties() {
+ return cubeDescOverrideProperties;
+ }
+
+ public void setCubeDescOverrideProperties(Map<String, String> cubeDescOverrideProperties) {
+ this.cubeDescOverrideProperties = cubeDescOverrideProperties;
+ }
+
+ public void setStreamingProperties(Map<String, String> streamingProperties) {
+ this.streamingProperties = streamingProperties;
+ }
+
+ public Map<String, String> getStreamingProperties() {
+ return streamingProperties;
+ }
+
+ public void init() {
+ streamingProperties.put("parserProperties",
+ "tsColName=" + RecordEvent.RecordReserveKeyEnum.TIME.toString() + ";");
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/74e167f0/tool/src/main/resources/SCSinkTools.json
----------------------------------------------------------------------
diff --git a/tool/src/main/resources/SCSinkTools.json b/tool/src/main/resources/SCSinkTools.json
new file mode 100644
index 0000000..15a715f
--- /dev/null
+++ b/tool/src/main/resources/SCSinkTools.json
@@ -0,0 +1,14 @@
+[
+ [
+ "org.apache.kylin.tool.metrics.systemcube.util.HiveSinkTool",
+ {
+ "storageType": 2,
+ "cubeDescOverrideProperties": [
+ "java.util.HashMap",
+ {
+ "kylin.cube.algorithm": "INMEM"
+ }
+ ]
+ }
+ ]
+]
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kylin/blob/74e167f0/tool/src/test/java/org/apache/kylin/tool/metrics/systemcube/SCCreatorTest.java
----------------------------------------------------------------------
diff --git a/tool/src/test/java/org/apache/kylin/tool/metrics/systemcube/SCCreatorTest.java b/tool/src/test/java/org/apache/kylin/tool/metrics/systemcube/SCCreatorTest.java
new file mode 100644
index 0000000..6718723
--- /dev/null
+++ b/tool/src/test/java/org/apache/kylin/tool/metrics/systemcube/SCCreatorTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.tool.metrics.systemcube;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.kylin.common.util.LocalFileMetadataTestCase;
+import org.apache.kylin.metadata.MetadataManager;
+import org.apache.kylin.tool.metrics.systemcube.util.HiveSinkTool;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+public class SCCreatorTest extends LocalFileMetadataTestCase {
+
+ @Before
+ public void setUp() throws Exception {
+ this.createTestMetadata();
+ MetadataManager.clearCache();
+ }
+
+ @After
+ public void after() throws Exception {
+ this.cleanupTestMetadata();
+ }
+
+ @Test
+ public void testExecute() throws Exception {
+ String outputPath = "../examples/system";
+ String inputPath = "src/main/resources/SCSinkTools.json";
+
+ SCCreator cli = new SCCreator();
+ cli.execute("ADMIN", outputPath, inputPath);
+ }
+
+ @Test
+ public void testWriteSinkToolsJson() throws Exception {
+ Map<String, String> cubeDescOverrideProperties = Maps.newHashMap();
+ cubeDescOverrideProperties.put("kylin.cube.algorithm", "INMEM");
+
+ HiveSinkTool hiveSinkTool = new HiveSinkTool();
+ hiveSinkTool.setCubeDescOverrideProperties(cubeDescOverrideProperties);
+
+ try (BufferedOutputStream os = new BufferedOutputStream(
+ new FileOutputStream("src/test/resources/SCSinkTools.json"))) {
+ ObjectMapper mapper = new ObjectMapper();
+ mapper.enableDefaultTyping();
+ mapper.writeValue(os, Sets.newHashSet(hiveSinkTool));
+ }
+ }
+
+ @Test
+ public void testReadSinkToolsJson() throws Exception {
+ try (BufferedInputStream is = new BufferedInputStream(
+ new FileInputStream("src/main/resources/SCSinkTools.json"))) {
+ ObjectMapper mapper = new ObjectMapper();
+ mapper.enableDefaultTyping();
+ Set<HiveSinkTool> sinkToolSet = mapper.readValue(is, HashSet.class);
+ for (HiveSinkTool entry : sinkToolSet) {
+ System.out.println(entry.getCubeDescOverrideProperties());
+ }
+ }
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kylin/blob/74e167f0/tool/src/test/resources/SCSinkTools.json
----------------------------------------------------------------------
diff --git a/tool/src/test/resources/SCSinkTools.json b/tool/src/test/resources/SCSinkTools.json
new file mode 100644
index 0000000..15a715f
--- /dev/null
+++ b/tool/src/test/resources/SCSinkTools.json
@@ -0,0 +1,14 @@
+[
+ [
+ "org.apache.kylin.tool.metrics.systemcube.util.HiveSinkTool",
+ {
+ "storageType": 2,
+ "cubeDescOverrideProperties": [
+ "java.util.HashMap",
+ {
+ "kylin.cube.algorithm": "INMEM"
+ }
+ ]
+ }
+ ]
+]
\ No newline at end of file