You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2015/01/23 09:32:22 UTC
[35/50] [abbrv] incubator-kylin git commit: Merge branch
'origin/staging'
Merge branch 'origin/staging'
Conflicts:
job/src/test/java/com/kylinolap/job/BuildCubeWithEngineTest.java
job/src/test/java/com/kylinolap/job/DeployUtil.java
job/src/test/java/com/kylinolap/job/SampleCubeSetupTest.java
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/29cbf339
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/29cbf339
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/29cbf339
Branch: refs/heads/inverted-index
Commit: 29cbf33923fd155353e53def3b1ef4774e3dcf76
Parents: 8da37ea 9feb25a
Author: honma <ho...@ebay.com>
Authored: Fri Jan 23 13:35:52 2015 +0800
Committer: honma <ho...@ebay.com>
Committed: Fri Jan 23 13:44:03 2015 +0800
----------------------------------------------------------------------
.../sandbox/kylin_job_conf.lzo_disabled.xml | 60 ---
.../sandbox/kylin_job_conf.lzo_enabled.xml | 60 ---
.../job/tools/LZOSupportnessChecker.java | 4 +
.../kylinolap/job/BuildCubeWithEngineTest.java | 2 +-
.../kylinolap/job/BuildIIWithEngineTest.java | 1 -
.../test/java/com/kylinolap/job/DeployUtil.java | 497 +++++++++----------
.../com/kylinolap/job/SampleCubeSetupTest.java | 4 -
kylin.sh | 41 ++
package.sh | 6 +-
sandbox_deploy.sh | 41 +-
webapp/app/js/controllers/cubeSchema.js | 8 +-
11 files changed, 310 insertions(+), 414 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29cbf339/job/src/main/java/com/kylinolap/job/tools/LZOSupportnessChecker.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29cbf339/job/src/test/java/com/kylinolap/job/BuildCubeWithEngineTest.java
----------------------------------------------------------------------
diff --cc job/src/test/java/com/kylinolap/job/BuildCubeWithEngineTest.java
index e39e80b,2d1ae5e..82781de
--- a/job/src/test/java/com/kylinolap/job/BuildCubeWithEngineTest.java
+++ b/job/src/test/java/com/kylinolap/job/BuildCubeWithEngineTest.java
@@@ -78,30 -65,16 +78,30 @@@ public class BuildCubeWithEngineTest
DeployUtil.initCliWorkDir();
DeployUtil.deployMetadata();
DeployUtil.overrideJobJarLocations();
- DeployUtil.overrideJobConf(HBaseMetadataTestCase.SANDBOX_TEST_DATA);
++
+
+ final KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
+ jobService = ExecutableManager.getInstance(kylinConfig);
+ scheduler = DefaultScheduler.getInstance();
+ scheduler.init(new JobEngineConfig(kylinConfig));
+ if (!scheduler.hasStarted()) {
+ throw new RuntimeException("scheduler has not been started");
+ }
+ cubeManager = CubeManager.getInstance(kylinConfig);
+ jobEngineConfig = new JobEngineConfig(kylinConfig);
+ for (String jobId: jobService.getAllJobIds()) {
+ jobService.deleteJob(jobId);
+ }
- engineConfig = new JobEngineConfig(KylinConfig.getInstanceFromEnv());
- jobManager = new JobManager("Build_Test_Cube_Engine", engineConfig);
- jobManager.deleteAllJobs();
}
- @After
- public void after() throws IOException {
- // jobManager.deleteAllJobs();
- this.cleanupTestMetadata();
+ private void backup() throws Exception {
+ int exitCode = cleanupOldCubes();
+ if (exitCode == 0) {
+ exportHBaseData();
+ }
+
+ HBaseMetadataTestCase.staticCleanupTestMetadata();
}
@Test
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29cbf339/job/src/test/java/com/kylinolap/job/BuildIIWithEngineTest.java
----------------------------------------------------------------------
diff --cc job/src/test/java/com/kylinolap/job/BuildIIWithEngineTest.java
index 612eb79,0000000..4663b28
mode 100644,000000..100644
--- a/job/src/test/java/com/kylinolap/job/BuildIIWithEngineTest.java
+++ b/job/src/test/java/com/kylinolap/job/BuildIIWithEngineTest.java
@@@ -1,205 -1,0 +1,204 @@@
+package com.kylinolap.job;
+
+import static org.junit.Assert.*;
+
+import java.io.File;
+import java.lang.reflect.Method;
+import java.text.SimpleDateFormat;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+import com.kylinolap.common.KylinConfig;
+import com.kylinolap.common.util.AbstractKylinTestCase;
+import com.kylinolap.common.util.ClasspathUtil;
+import com.kylinolap.common.util.HBaseMetadataTestCase;
+import com.kylinolap.invertedindex.IIInstance;
+import com.kylinolap.invertedindex.IIManager;
+import com.kylinolap.invertedindex.IISegment;
+import com.kylinolap.job.engine.JobEngineConfig;
+import com.kylinolap.job.execution.ExecutableState;
+import com.kylinolap.job.impl.threadpool.AbstractExecutable;
+import com.kylinolap.job.impl.threadpool.DefaultScheduler;
+import com.kylinolap.job.invertedindex.IIJob;
+import com.kylinolap.job.invertedindex.IIJobBuilder;
+import com.kylinolap.job.service.ExecutableManager;
+import com.kylinolap.metadata.realization.RealizationStatusEnum;
+
+/**
+ *
+ * @author shaoshi
+ *
+ */
+public class BuildIIWithEngineTest {
+
+ private JobEngineConfig jobEngineConfig;
+
+ private IIManager iiManager;
+
+ private DefaultScheduler scheduler;
+
+ protected ExecutableManager jobService;
+
+ protected static final String TEST_II_NAME = "test_kylin_ii";
+
+ protected void waitForJob(String jobId) {
+ while (true) {
+ AbstractExecutable job = jobService.getJob(jobId);
+ if (job.getStatus() == ExecutableState.SUCCEED || job.getStatus() == ExecutableState.ERROR) {
+ break;
+ } else {
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ ClasspathUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ System.setProperty("hdp.version", "2.2.0.0-2041"); // mapred-site.xml ref this
+ }
+
+ @Before
+ public void before() throws Exception {
+ HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
+
+ DeployUtil.initCliWorkDir();
+// DeployUtil.deployMetadata();
+ DeployUtil.overrideJobJarLocations();
- DeployUtil.overrideJobConf(HBaseMetadataTestCase.SANDBOX_TEST_DATA);
+
+ final KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
+ jobService = ExecutableManager.getInstance(kylinConfig);
+ scheduler = DefaultScheduler.getInstance();
+ scheduler.init(new JobEngineConfig(kylinConfig));
+ if (!scheduler.hasStarted()) {
+ throw new RuntimeException("scheduler has not been started");
+ }
+ iiManager = IIManager.getInstance(kylinConfig);
+ jobEngineConfig = new JobEngineConfig(kylinConfig);
+ for (String jobId: jobService.getAllJobIds()) {
+ jobService.deleteJob(jobId);
+ }
+
+ IIInstance ii = iiManager.getII(TEST_II_NAME);
+ if(ii.getStatus() != RealizationStatusEnum.DISABLED) {
+ ii.setStatus(RealizationStatusEnum.DISABLED);
+ iiManager.updateII(ii);
+ }
+
+ }
+
+
+ @Test
+ private void testBuildII() throws Exception {
+
+ String[] testCase = new String[]{
+ "buildII"
+ };
+ ExecutorService executorService = Executors.newFixedThreadPool(testCase.length);
+ final CountDownLatch countDownLatch = new CountDownLatch(testCase.length);
+ List<Future<List<String>>> tasks = Lists.newArrayListWithExpectedSize(testCase.length);
+ for (int i = 0; i < testCase.length; i++) {
+ tasks.add(executorService.submit(new TestCallable(testCase[i], countDownLatch)));
+ }
+ countDownLatch.await();
+ for (int i = 0; i < tasks.size(); ++i) {
+ Future<List<String>> task = tasks.get(i);
+ final List<String> jobIds = task.get();
+ for (String jobId: jobIds) {
+ assertJobSucceed(jobId);
+ }
+ }
+ }
+
+
+ private void assertJobSucceed(String jobId) {
+ assertEquals(ExecutableState.SUCCEED, jobService.getOutput(jobId).getState());
+ }
+
+ private class TestCallable implements Callable<List<String>> {
+
+ private final String methodName;
+ private final CountDownLatch countDownLatch;
+
+ public TestCallable(String methodName, CountDownLatch countDownLatch) {
+ this.methodName = methodName;
+ this.countDownLatch = countDownLatch;
+ }
+
+ @Override
+ public List<String> call() throws Exception {
+ try {
+ final Method method = BuildIIWithEngineTest.class.getDeclaredMethod(methodName);
+ method.setAccessible(true);
+ return (List<String>) method.invoke(BuildIIWithEngineTest.this);
+ } finally {
+ countDownLatch.countDown();
+ }
+ }
+ }
+
+
+ protected List<String> buildII() throws Exception {
+ clearSegment(TEST_II_NAME);
+
+
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+
+ // this cube's start date is 0, end date is 20501112000000
+ long date1 = 0;
+ long date2 = f.parse("2013-01-01").getTime();
+
+
+ // this cube doesn't support incremental build, always do full build
+
+ List<String> result = Lists.newArrayList();
+ result.add(buildSegment(TEST_II_NAME, date1, date2));
+ return result;
+ }
+
+
+
+ private void clearSegment(String iiName) throws Exception{
+ IIInstance ii = iiManager.getII(iiName);
+ ii.getSegments().clear();
+ iiManager.updateII(ii);
+ }
+
+
+ private String buildSegment(String iiName, long startDate, long endDate) throws Exception {
+ IIInstance iiInstance = iiManager.getII(iiName);
+ IISegment segment = iiManager.buildSegment(iiInstance, startDate, endDate);
+ iiInstance.getSegments().add(segment);
+ iiManager.updateII(iiInstance);
+ IIJobBuilder iiJobBuilder = (IIJobBuilder)IIJobBuilder.newBuilder().setJobEnginConfig(jobEngineConfig).setSegment(segment);
+ IIJob job = iiJobBuilder.buildJob();
+ jobService.addJob(job);
+ waitForJob(job.getId());
+ return job.getId();
+ }
+
+ public static void main(String[] args) throws Exception {
+ BuildIIWithEngineTest instance = new BuildIIWithEngineTest();
+
+ BuildIIWithEngineTest.beforeClass();
+ instance.before();
+ instance.testBuildII();
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29cbf339/job/src/test/java/com/kylinolap/job/DeployUtil.java
----------------------------------------------------------------------
diff --cc job/src/test/java/com/kylinolap/job/DeployUtil.java
index f1929ed,ecedfda..6794aa9
--- a/job/src/test/java/com/kylinolap/job/DeployUtil.java
+++ b/job/src/test/java/com/kylinolap/job/DeployUtil.java
@@@ -1,254 -1,205 +1,243 @@@
--package com.kylinolap.job;
--
--import java.io.File;
--import java.io.FileOutputStream;
--import java.io.FileReader;
--import java.io.IOException;
--import java.io.InputStream;
--
--import org.apache.commons.io.IOUtils;
--import org.apache.hadoop.hbase.util.Pair;
--import org.apache.maven.model.Model;
--import org.apache.maven.model.io.xpp3.MavenXpp3Reader;
- import org.apache.tools.ant.filters.StringInputStream;
--import org.codehaus.plexus.util.FileUtils;
--import org.slf4j.Logger;
--import org.slf4j.LoggerFactory;
--
--import com.kylinolap.common.KylinConfig;
- import com.kylinolap.common.persistence.ResourceStore;
--import com.kylinolap.common.persistence.ResourceTool;
--import com.kylinolap.common.util.AbstractKylinTestCase;
--import com.kylinolap.common.util.CliCommandExecutor;
- import com.kylinolap.common.util.HiveClient;
--import com.kylinolap.cube.CubeInstance;
--import com.kylinolap.cube.CubeManager;
- import com.kylinolap.job.dataGen.FactTableGenerator;
-import com.kylinolap.cube.dataGen.FactTableGenerator;
--import com.kylinolap.job.engine.JobEngineConfig;
--import com.kylinolap.job.hadoop.hive.SqlHiveDataTypeMapping;
--import com.kylinolap.job.tools.LZOSupportnessChecker;
--import com.kylinolap.metadata.MetadataManager;
- import com.kylinolap.metadata.model.ColumnDesc;
- import com.kylinolap.metadata.model.TableDesc;
-import com.kylinolap.metadata.model.schema.ColumnDesc;
-import com.kylinolap.metadata.model.schema.TableDesc;
--
--public class DeployUtil {
-- @SuppressWarnings("unused")
-- private static final Logger logger = LoggerFactory.getLogger(DeployUtil.class);
--
-- public static void initCliWorkDir() throws IOException {
-- execCliCommand("rm -rf " + getHadoopCliWorkingDir());
-- execCliCommand("mkdir -p " + config().getKylinJobLogDir());
-- }
--
-- public static void deployMetadata() throws IOException {
-- // install metadata to hbase
-- ResourceTool.reset(config());
-- ResourceTool.copy(KylinConfig.createInstanceFromUri(AbstractKylinTestCase.LOCALMETA_TEST_DATA), config());
--
-- // update cube desc signature.
-- for (CubeInstance cube : CubeManager.getInstance(config()).listAllCubes()) {
-- cube.getDescriptor().setSignature(cube.getDescriptor().calculateSignature());
-- CubeManager.getInstance(config()).updateCube(cube);
-- }
-- }
--
-- public static void overrideJobJarLocations() {
-- Pair<File, File> files = getJobJarFiles();
-- File jobJar = files.getFirst();
-- File coprocessorJar = files.getSecond();
--
-- config().overrideKylinJobJarPath(jobJar.getAbsolutePath());
-- config().overrideCoprocessorLocalJar(coprocessorJar.getAbsolutePath());
-- }
-
-
-- public static void deployJobJars() throws IOException {
-- Pair<File, File> files = getJobJarFiles();
-- File originalJobJar = files.getFirst();
-- File originalCoprocessorJar = files.getSecond();
--
-- File targetJobJar = new File(config().getKylinJobJarPath());
-- File jobJarRenamedAsTarget = new File(originalJobJar.getParentFile(), targetJobJar.getName());
-- if (originalJobJar.equals(jobJarRenamedAsTarget) == false) {
-- FileUtils.copyFile(originalJobJar, jobJarRenamedAsTarget);
-- }
--
-- File targetCoprocessorJar = new File(config().getCoprocessorLocalJar());
-- File coprocessorJarRenamedAsTarget = new File(originalCoprocessorJar.getParentFile(), targetCoprocessorJar.getName());
-- if (originalCoprocessorJar.equals(coprocessorJarRenamedAsTarget) == false) {
-- FileUtils.copyFile(originalCoprocessorJar, coprocessorJarRenamedAsTarget);
-- }
-
-
-- CliCommandExecutor cmdExec = config().getCliCommandExecutor();
-- cmdExec.copyFile(jobJarRenamedAsTarget.getAbsolutePath(), targetJobJar.getParent());
-- cmdExec.copyFile(coprocessorJarRenamedAsTarget.getAbsolutePath(), targetCoprocessorJar.getParent());
-- }
-
-
-- private static Pair<File, File> getJobJarFiles() {
-- String version;
-- try {
-- MavenXpp3Reader pomReader = new MavenXpp3Reader();
-- Model model = pomReader.read(new FileReader("../pom.xml"));
-- version = model.getVersion();
-- } catch (Exception e) {
-- throw new RuntimeException(e.getMessage(), e);
-- }
--
-- File jobJar = new File("../job/target", "kylin-job-" + version + "-job.jar");
-- File coprocessorJar = new File("../storage/target", "kylin-storage-" + version + "-coprocessor.jar");
-- return new Pair<File, File>(jobJar, coprocessorJar);
- }
-
- public static void overrideJobConf(String confDir) throws IOException {
- boolean enableLzo = LZOSupportnessChecker.getSupportness();
- overrideJobConf(confDir, enableLzo);
- }
-
- public static void overrideJobConf(String confDir, boolean enableLzo) throws IOException {
- File src = new File(confDir, JobEngineConfig.HADOOP_JOB_CONF_FILENAME + (enableLzo ? ".lzo_enabled" : ".lzo_disabled") + ".xml");
- File dst = new File(confDir, JobEngineConfig.HADOOP_JOB_CONF_FILENAME + ".xml");
- FileUtils.copyFile(src, dst);
-- }
-
-
-- private static void execCliCommand(String cmd) throws IOException {
-- config().getCliCommandExecutor().execute(cmd);
-- }
--
-- private static String getHadoopCliWorkingDir() {
-- return config().getCliWorkingDir();
-- }
--
-- private static KylinConfig config() {
-- return KylinConfig.getInstanceFromEnv();
-- }
--
-- // ============================================================================
--
- static final String TABLE_CAL_DT = "edw.test_cal_dt";
- static final String TABLE_CATEGORY_GROUPINGS = "default.test_category_groupings";
- static final String TABLE_KYLIN_FACT = "default.test_kylin_fact";
- static final String TABLE_SELLER_TYPE_DIM = "edw.test_seller_type_dim";
- static final String TABLE_SITES = "edw.test_sites";
- static final String TABLE_CAL_DT = "test_cal_dt";
- static final String TABLE_CATEGORY_GROUPINGS = "test_category_groupings";
- static final String TABLE_KYLIN_FACT = "test_kylin_fact";
- static final String TABLE_SELLER_TYPE_DIM = "test_seller_type_dim";
- static final String TABLE_SITES = "test_sites";
--
-- static final String[] TABLE_NAMES = new String[] { TABLE_CAL_DT, TABLE_CATEGORY_GROUPINGS, TABLE_KYLIN_FACT, TABLE_SELLER_TYPE_DIM, TABLE_SITES };
--
-- public static void prepareTestData(String joinType, String cubeName) throws Exception {
-
- String factTableName = TABLE_KYLIN_FACT.toUpperCase();
- String content = null;
-
- boolean buildCubeUsingProvidedData = Boolean.parseBoolean(System.getProperty("buildCubeUsingProvidedData"));
- if (!buildCubeUsingProvidedData) {
- System.out.println("build cube with random dataset");
- // data is generated according to cube descriptor and saved in resource store
- if (joinType.equalsIgnoreCase("inner")) {
- content = FactTableGenerator.generate(cubeName, "10000", "1", null, "inner");
- } else if (joinType.equalsIgnoreCase("left")) {
- content = FactTableGenerator.generate(cubeName, "10000", "0.6", null, "left");
- } else {
- throw new IllegalArgumentException("Unsupported join type : " + joinType);
- }
-
- assert content != null;
- overrideFactTableData(content, factTableName);
- // data is generated according to cube descriptor and saved in resource store
- if (joinType.equalsIgnoreCase("inner")) {
- FactTableGenerator.generate(cubeName, "10000", "1", null, "inner");
- } else if (joinType.equalsIgnoreCase("left")) {
- FactTableGenerator.generate(cubeName, "10000", "0.6", null, "left");
-- } else {
- System.out.println("build cube with provided dataset");
- throw new IllegalArgumentException("Unsupported join type : " + joinType);
-- }
--
- duplicateFactTableData(factTableName, joinType);
-- deployHiveTables();
- }
-
- public static void overrideFactTableData(String factTableContent, String factTableName) throws IOException {
- // Write to resource store
- ResourceStore store = ResourceStore.getStore(config());
-
- InputStream in = new StringInputStream(factTableContent);
- String factTablePath = "/data/" + factTableName + ".csv";
- store.deleteResource(factTablePath);
- store.putResource(factTablePath, in, System.currentTimeMillis());
- in.close();
- }
-
- public static void duplicateFactTableData(String factTableName, String joinType) throws IOException {
- // duplicate a copy of this fact table, with a naming convention with fact.csv.inner or fact.csv.left
- // so that later test cases can select different data files
- ResourceStore store = ResourceStore.getStore(config());
- InputStream in = store.getResource("/data/" + factTableName + ".csv");
- String factTablePathWithJoinType = "/data/" + factTableName + ".csv." + joinType.toLowerCase();
- store.deleteResource(factTablePathWithJoinType);
- store.putResource(factTablePathWithJoinType, in, System.currentTimeMillis());
- in.close();
-- }
--
-- private static void deployHiveTables() throws Exception {
--
-- MetadataManager metaMgr = MetadataManager.getInstance(config());
--
-- // scp data files, use the data from hbase, instead of local files
-- File temp = File.createTempFile("temp", ".csv");
-- temp.createNewFile();
-- for (String tablename : TABLE_NAMES) {
-- tablename = tablename.toUpperCase();
--
-- File localBufferFile = new File(temp.getParent() + "/" + tablename + ".csv");
-- localBufferFile.createNewFile();
--
-- InputStream hbaseDataStream = metaMgr.getStore().getResource("/data/" + tablename + ".csv");
-- FileOutputStream localFileStream = new FileOutputStream(localBufferFile);
-- IOUtils.copy(hbaseDataStream, localFileStream);
--
-- hbaseDataStream.close();
-- localFileStream.close();
--
- localBufferFile.deleteOnExit();
- config().getCliCommandExecutor().copyFile(localBufferFile.getPath(), config().getCliWorkingDir());
- localBufferFile.delete();
-- }
- String tableFileDir = temp.getParent();
-- temp.delete();
--
- HiveClient hiveClient = new HiveClient();
-
-- // create hive tables
- hiveClient.executeHQL("CREATE DATABASE IF NOT EXISTS EDW");
- hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_CAL_DT.toUpperCase())));
- hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_CATEGORY_GROUPINGS.toUpperCase())));
- hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_KYLIN_FACT.toUpperCase())));
- hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_SELLER_TYPE_DIM.toUpperCase())));
- hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_SITES.toUpperCase())));
- execHiveCommand(generateCreateTableHql(metaMgr.getTableDesc(TABLE_CAL_DT.toUpperCase())));
- execHiveCommand(generateCreateTableHql(metaMgr.getTableDesc(TABLE_CATEGORY_GROUPINGS.toUpperCase())));
- execHiveCommand(generateCreateTableHql(metaMgr.getTableDesc(TABLE_KYLIN_FACT.toUpperCase())));
- execHiveCommand(generateCreateTableHql(metaMgr.getTableDesc(TABLE_SELLER_TYPE_DIM.toUpperCase())));
- execHiveCommand(generateCreateTableHql(metaMgr.getTableDesc(TABLE_SITES.toUpperCase())));
--
-- // load data to hive tables
-- // LOAD DATA LOCAL INPATH 'filepath' [OVERWRITE] INTO TABLE tablename
- hiveClient.executeHQL(generateLoadDataHql(TABLE_CAL_DT, tableFileDir));
- hiveClient.executeHQL(generateLoadDataHql(TABLE_CATEGORY_GROUPINGS, tableFileDir));
- hiveClient.executeHQL(generateLoadDataHql(TABLE_KYLIN_FACT, tableFileDir));
- hiveClient.executeHQL(generateLoadDataHql(TABLE_SELLER_TYPE_DIM, tableFileDir));
- hiveClient.executeHQL(generateLoadDataHql(TABLE_SITES, tableFileDir));
- execHiveCommand(generateLoadDataHql(TABLE_CAL_DT));
- execHiveCommand(generateLoadDataHql(TABLE_CATEGORY_GROUPINGS));
- execHiveCommand(generateLoadDataHql(TABLE_KYLIN_FACT));
- execHiveCommand(generateLoadDataHql(TABLE_SELLER_TYPE_DIM));
- execHiveCommand(generateLoadDataHql(TABLE_SITES));
-- }
--
- private static String generateLoadDataHql(String tableName, String tableFileDir) {
- return "LOAD DATA LOCAL INPATH '" + tableFileDir + "/" + tableName.toUpperCase() + ".csv' OVERWRITE INTO TABLE " + tableName.toUpperCase();
- private static void execHiveCommand(String hql) throws IOException {
- String hiveCmd = "hive -e \"" + hql + "\"";
- config().getCliCommandExecutor().execute(hiveCmd);
-- }
--
- private static String[] generateCreateTableHql(TableDesc tableDesc) {
-
- String dropsql = "DROP TABLE IF EXISTS " + tableDesc.getIdentity();
- private static String generateLoadDataHql(String tableName) {
- return "LOAD DATA LOCAL INPATH '" + config().getCliWorkingDir() + "/" + tableName.toUpperCase() + ".csv' OVERWRITE INTO TABLE " + tableName.toUpperCase();
- }
-
- private static String generateCreateTableHql(TableDesc tableDesc) {
-- StringBuilder ddl = new StringBuilder();
--
- ddl.append("CREATE TABLE " + tableDesc.getIdentity() + "\n");
- ddl.append("DROP TABLE IF EXISTS " + tableDesc.getName() + ";\n");
- ddl.append("CREATE TABLE " + tableDesc.getName() + "\n");
-- ddl.append("(" + "\n");
--
-- for (int i = 0; i < tableDesc.getColumns().length; i++) {
-- ColumnDesc col = tableDesc.getColumns()[i];
-- if (i > 0) {
-- ddl.append(",");
-- }
-- ddl.append(col.getName() + " " + SqlHiveDataTypeMapping.getHiveDataType((col.getDatatype())) + "\n");
-- }
--
-- ddl.append(")" + "\n");
-- ddl.append("ROW FORMAT DELIMITED FIELDS TERMINATED BY ','" + "\n");
- ddl.append("STORED AS TEXTFILE");
- ddl.append("STORED AS TEXTFILE;");
--
- return new String[] {dropsql, ddl.toString()};
- return ddl.toString();
-- }
--
--}
++package com.kylinolap.job;
++
++import java.io.File;
++import java.io.FileOutputStream;
++import java.io.FileReader;
++import java.io.IOException;
++import java.io.InputStream;
++
++import org.apache.commons.io.IOUtils;
++import org.apache.hadoop.hbase.util.Pair;
++import org.apache.maven.model.Model;
++import org.apache.maven.model.io.xpp3.MavenXpp3Reader;
++import org.apache.tools.ant.filters.StringInputStream;
++import org.codehaus.plexus.util.FileUtils;
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
++
++import com.kylinolap.common.KylinConfig;
++import com.kylinolap.common.persistence.ResourceStore;
++import com.kylinolap.common.persistence.ResourceTool;
++import com.kylinolap.common.util.AbstractKylinTestCase;
++import com.kylinolap.common.util.CliCommandExecutor;
++import com.kylinolap.common.util.HiveClient;
++import com.kylinolap.cube.CubeInstance;
++import com.kylinolap.cube.CubeManager;
++import com.kylinolap.job.dataGen.FactTableGenerator;
++import com.kylinolap.job.engine.JobEngineConfig;
++import com.kylinolap.job.hadoop.hive.SqlHiveDataTypeMapping;
++import com.kylinolap.job.tools.LZOSupportnessChecker;
++import com.kylinolap.metadata.MetadataManager;
++import com.kylinolap.metadata.model.ColumnDesc;
++import com.kylinolap.metadata.model.TableDesc;
++
++public class DeployUtil {
++ @SuppressWarnings("unused")
++ private static final Logger logger = LoggerFactory.getLogger(DeployUtil.class);
++
++ public static void initCliWorkDir() throws IOException {
++ execCliCommand("rm -rf " + getHadoopCliWorkingDir());
++ execCliCommand("mkdir -p " + config().getKylinJobLogDir());
++ }
++
++ public static void deployMetadata() throws IOException {
++ // install metadata to hbase
++ ResourceTool.reset(config());
++ ResourceTool.copy(KylinConfig.createInstanceFromUri(AbstractKylinTestCase.LOCALMETA_TEST_DATA), config());
++
++ // update cube desc signature.
++ for (CubeInstance cube : CubeManager.getInstance(config()).listAllCubes()) {
++ cube.getDescriptor().setSignature(cube.getDescriptor().calculateSignature());
++ CubeManager.getInstance(config()).updateCube(cube);
++ }
++ }
++
++ public static void overrideJobJarLocations() {
++ Pair<File, File> files = getJobJarFiles();
++ File jobJar = files.getFirst();
++ File coprocessorJar = files.getSecond();
++
++ config().overrideKylinJobJarPath(jobJar.getAbsolutePath());
++ config().overrideCoprocessorLocalJar(coprocessorJar.getAbsolutePath());
++ }
++
++ public static void deployJobJars() throws IOException {
++ Pair<File, File> files = getJobJarFiles();
++ File originalJobJar = files.getFirst();
++ File originalCoprocessorJar = files.getSecond();
++
++ File targetJobJar = new File(config().getKylinJobJarPath());
++ File jobJarRenamedAsTarget = new File(originalJobJar.getParentFile(), targetJobJar.getName());
++ if (originalJobJar.equals(jobJarRenamedAsTarget) == false) {
++ FileUtils.copyFile(originalJobJar, jobJarRenamedAsTarget);
++ }
++
++ File targetCoprocessorJar = new File(config().getCoprocessorLocalJar());
++ File coprocessorJarRenamedAsTarget = new File(originalCoprocessorJar.getParentFile(), targetCoprocessorJar.getName());
++ if (originalCoprocessorJar.equals(coprocessorJarRenamedAsTarget) == false) {
++ FileUtils.copyFile(originalCoprocessorJar, coprocessorJarRenamedAsTarget);
++ }
++
++ CliCommandExecutor cmdExec = config().getCliCommandExecutor();
++ cmdExec.copyFile(jobJarRenamedAsTarget.getAbsolutePath(), targetJobJar.getParent());
++ cmdExec.copyFile(coprocessorJarRenamedAsTarget.getAbsolutePath(), targetCoprocessorJar.getParent());
++ }
++
++ private static Pair<File, File> getJobJarFiles() {
++ String version;
++ try {
++ MavenXpp3Reader pomReader = new MavenXpp3Reader();
++ Model model = pomReader.read(new FileReader("../pom.xml"));
++ version = model.getVersion();
++ } catch (Exception e) {
++ throw new RuntimeException(e.getMessage(), e);
++ }
++
++ File jobJar = new File("../job/target", "kylin-job-" + version + "-job.jar");
++ File coprocessorJar = new File("../storage/target", "kylin-storage-" + version + "-coprocessor.jar");
++ return new Pair<File, File>(jobJar, coprocessorJar);
++ }
++
++ private static void execCliCommand(String cmd) throws IOException {
++ config().getCliCommandExecutor().execute(cmd);
++ }
++
++ private static String getHadoopCliWorkingDir() {
++ return config().getCliWorkingDir();
++ }
++
++ private static KylinConfig config() {
++ return KylinConfig.getInstanceFromEnv();
++ }
++
++ // ============================================================================
++
++ static final String TABLE_CAL_DT = "edw.test_cal_dt";
++ static final String TABLE_CATEGORY_GROUPINGS = "default.test_category_groupings";
++ static final String TABLE_KYLIN_FACT = "default.test_kylin_fact";
++ static final String TABLE_SELLER_TYPE_DIM = "edw.test_seller_type_dim";
++ static final String TABLE_SITES = "edw.test_sites";
++
++ static final String[] TABLE_NAMES = new String[] { TABLE_CAL_DT, TABLE_CATEGORY_GROUPINGS, TABLE_KYLIN_FACT, TABLE_SELLER_TYPE_DIM, TABLE_SITES };
++
++ public static void prepareTestData(String joinType, String cubeName) throws Exception {
++
++ String factTableName = TABLE_KYLIN_FACT.toUpperCase();
++ String content = null;
++
++ boolean buildCubeUsingProvidedData = Boolean.parseBoolean(System.getProperty("buildCubeUsingProvidedData"));
++ if (!buildCubeUsingProvidedData) {
++ System.out.println("build cube with random dataset");
++ // data is generated according to cube descriptor and saved in resource store
++ if (joinType.equalsIgnoreCase("inner")) {
++ content = FactTableGenerator.generate(cubeName, "10000", "1", null, "inner");
++ } else if (joinType.equalsIgnoreCase("left")) {
++ content = FactTableGenerator.generate(cubeName, "10000", "0.6", null, "left");
++ } else {
++ throw new IllegalArgumentException("Unsupported join type : " + joinType);
++ }
++
++ assert content != null;
++ overrideFactTableData(content, factTableName);
++ } else {
++ System.out.println("build cube with provided dataset");
++ }
++
++ duplicateFactTableData(factTableName, joinType);
++ deployHiveTables();
++ }
++
++ public static void overrideFactTableData(String factTableContent, String factTableName) throws IOException {
++ // Write to resource store
++ ResourceStore store = ResourceStore.getStore(config());
++
++ InputStream in = new StringInputStream(factTableContent);
++ String factTablePath = "/data/" + factTableName + ".csv";
++ store.deleteResource(factTablePath);
++ store.putResource(factTablePath, in, System.currentTimeMillis());
++ in.close();
++ }
++
++ public static void duplicateFactTableData(String factTableName, String joinType) throws IOException {
++ // duplicate a copy of this fact table, with a naming convention with fact.csv.inner or fact.csv.left
++ // so that later test cases can select different data files
++ ResourceStore store = ResourceStore.getStore(config());
++ InputStream in = store.getResource("/data/" + factTableName + ".csv");
++ String factTablePathWithJoinType = "/data/" + factTableName + ".csv." + joinType.toLowerCase();
++ store.deleteResource(factTablePathWithJoinType);
++ store.putResource(factTablePathWithJoinType, in, System.currentTimeMillis());
++ in.close();
++ }
++
++ private static void deployHiveTables() throws Exception {
++
++ MetadataManager metaMgr = MetadataManager.getInstance(config());
++
++ // scp data files, use the data from hbase, instead of local files
++ File temp = File.createTempFile("temp", ".csv");
++ temp.createNewFile();
++ for (String tablename : TABLE_NAMES) {
++ tablename = tablename.toUpperCase();
++
++ File localBufferFile = new File(temp.getParent() + "/" + tablename + ".csv");
++ localBufferFile.createNewFile();
++
++ InputStream hbaseDataStream = metaMgr.getStore().getResource("/data/" + tablename + ".csv");
++ FileOutputStream localFileStream = new FileOutputStream(localBufferFile);
++ IOUtils.copy(hbaseDataStream, localFileStream);
++
++ hbaseDataStream.close();
++ localFileStream.close();
++
++ localBufferFile.deleteOnExit();
++ }
++ String tableFileDir = temp.getParent();
++ temp.delete();
++
++ HiveClient hiveClient = new HiveClient();
++
++ // create hive tables
++ hiveClient.executeHQL("CREATE DATABASE IF NOT EXISTS EDW");
++ hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_CAL_DT.toUpperCase())));
++ hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_CATEGORY_GROUPINGS.toUpperCase())));
++ hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_KYLIN_FACT.toUpperCase())));
++ hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_SELLER_TYPE_DIM.toUpperCase())));
++ hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_SITES.toUpperCase())));
++
++ // load data to hive tables
++ // LOAD DATA LOCAL INPATH 'filepath' [OVERWRITE] INTO TABLE tablename
++ hiveClient.executeHQL(generateLoadDataHql(TABLE_CAL_DT, tableFileDir));
++ hiveClient.executeHQL(generateLoadDataHql(TABLE_CATEGORY_GROUPINGS, tableFileDir));
++ hiveClient.executeHQL(generateLoadDataHql(TABLE_KYLIN_FACT, tableFileDir));
++ hiveClient.executeHQL(generateLoadDataHql(TABLE_SELLER_TYPE_DIM, tableFileDir));
++ hiveClient.executeHQL(generateLoadDataHql(TABLE_SITES, tableFileDir));
++ }
++
++ private static String generateLoadDataHql(String tableName, String tableFileDir) {
++ return "LOAD DATA LOCAL INPATH '" + tableFileDir + "/" + tableName.toUpperCase() + ".csv' OVERWRITE INTO TABLE " + tableName.toUpperCase();
++ }
++
++ private static String[] generateCreateTableHql(TableDesc tableDesc) {
++
++ String dropsql = "DROP TABLE IF EXISTS " + tableDesc.getIdentity();
++ StringBuilder ddl = new StringBuilder();
++
++ ddl.append("CREATE TABLE " + tableDesc.getIdentity() + "\n");
++ ddl.append("(" + "\n");
++
++ for (int i = 0; i < tableDesc.getColumns().length; i++) {
++ ColumnDesc col = tableDesc.getColumns()[i];
++ if (i > 0) {
++ ddl.append(",");
++ }
++ ddl.append(col.getName() + " " + SqlHiveDataTypeMapping.getHiveDataType((col.getDatatype())) + "\n");
++ }
++
++ ddl.append(")" + "\n");
++ ddl.append("ROW FORMAT DELIMITED FIELDS TERMINATED BY ','" + "\n");
++ ddl.append("STORED AS TEXTFILE");
++
++ return new String[] {dropsql, ddl.toString()};
++ }
++
++}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29cbf339/job/src/test/java/com/kylinolap/job/SampleCubeSetupTest.java
----------------------------------------------------------------------